spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Steve Lewis <>
Subject A sample for generating big data - and some design questions
Date Wed, 01 Oct 2014 00:16:51 GMT
This sample below is essentially word count modified to be big data by
turning lines into groups of
upper case letters and then generating all case variants - it is modeled
after some real problems in biology
The issue is I know how to do this in Hadoop but in Spark the use of a List
in my flatmap function will not work as the size grows but I don't know
what will or how not to keep data in memory
Anyone want to look at the sample and tell me how

on my machine given 8g it does
Variant Size 18 Size 14188672 took 406 sec and
stalls with larger cases
import org.apache.spark.*;
import scala.*;

import java.util.*;

 * com.lordjoe.distributed.test.JavaBigDataWordCount
 * This sample is written to force a sample with large amounts of data
emulating some big data aspects of
 * a problem in biology I am working on -
 * <p/>
 * This is essentially WordCount
 * except that lines are filtered to just upper case words
 * then broken into String groups and all varients with different case are
 * so THE -> THE,ThE,THe,The,tHE,thE,tHe,the
 * when the groups get long - say 10 or 20 a LOT of variants are generated
 * <p/>
 * This sample is motivated by real problems in biology where we want to
look at possible mutations in DNA fragments or
 * possible chemical modifications on amino acids in polypeptides - my
largest Hadoop job does exactly that
 * <p/>
 * I have serious questions about
 * A - How to write the FlatMapFunction CaseVariationFunction as the output
gets large - I think storing results in a List will not work
 * - what are other options
 * B are there other ways to do this
public final class JavaBigDataWordCount {

     * drop all characters that are not letters
     * @param s input string
     * @return output string
    public static String dropNonLetters(String s) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < s.length(); i++) {
            char c = s.charAt(i);
            if (Character.isLetter(c))

        return sb.toString();

     * convert a string into a string holding only upper case letters
     * @param inp input string
     * @return output string
    public static String regularizeString(String inp) {
        inp = inp.trim();
        inp = inp.toUpperCase();
        return dropNonLetters(inp);

     * convert a string into strings of length maxLength all letters and
     * upper case
    public static class SubstringsMapFunction implements
FlatMapFunction<String, String> {
        private final int maxLength;

        public SubstringsMapFunction(final int pMaxLength) {
            maxLength = pMaxLength;

        public Iterable<String> call(String s) {
            s = regularizeString(s); // drop non-letters
            List<String> holder = new ArrayList<String>();
            for (int i = 0; i < s.length() - 2; i += maxLength) {
                holder.add(s.substring(i, Math.min(s.length(), i +
            return holder;

     * return all cases of an upper case string so THE ->
     * In general the output is 2 to the Nth long where N is the input
    public static class CaseVariationFunction implements
FlatMapFunction<String, String> {
        public Iterable<String> call(String s) {
            // HELP - I don't think a List will work for long inputs given
WHAT else can I use
            List<String> holder = new ArrayList<String>();  // holds
            makeVariations(s.toCharArray(), holder, 0);  // do real work
filling in holder
            return holder;

         * add to holder - NOTE I think a List is wrong for large inputs
         * @param chars  characters input
         * @param holder - holder - or iterable holding results
         * @param index  - start changing case at this position
        private void makeVariations(char[] chars, final List<String>
holder, int index) {
            if (index < chars.length - 1)
                makeVariations(chars, holder, index + 1);
            if (Character.isUpperCase(chars[index])) {
                chars[index] = Character.toLowerCase(chars[index]);
                holder.add(new String(chars));
                if (index < chars.length - 1)
                    makeVariations(chars, holder, index + 1);
                chars[index] = Character.toUpperCase(chars[index]);


    // a few lines of text so we don't need to read a file is we don't want
    public static final String GETTYSBURG =
            "Four score and seven years ago our fathers brought forth, upon
this continent, a new nation, conceived in liberty," +
                    " and dedicated to the proposition that “all men are
created equal.”\n" +
                    "Now we are engaged in a great civil war, testing
whether that nation, or any nation so conceived," +
                    " and so dedicated, can long endure. We are met on a
great battle field of that war. We come to dedicate a portion of it," +
                    " as a final resting place for those who died here,
that the nation might live. This we may, in all propriety do.\n" +
                    "But, in a larger sense, we can not dedicate – we can
not consecrate – we can not hallow, this ground – The brave men, living and
dead, who struggled here, have hallowed it, far above our poor power to add
or detract. The world will little note, nor long remember what we say here;
while it can never forget what they did here.\n" +
                    "It is rather for us, the living, we here be dedicated
to the great task remaining before us – that, from these honored dead" +
                    " we take increased devotion to that cause for which
they here, gave the last full measure of devotion – that we here highly
resolve" +
                    " these dead shall not have died in vain; that the
nation, shall have a new birth of freedom, and that government of the
people, by" +
                    " the people, for the people, shall not perish from the

     * Main - runs essentially word count for all variants starting with
length 10 (1000 variants) and
     * raising it by a factor of 4
     * @param args if given a file to read - otherwise use the Gettysburg
     * @throws Exception
    public static void main(String[] args) throws Exception {
        SparkConf sparkConf = new
        sparkConf.set("spark.mesos.coarse", "true");
        Option<String> option = sparkConf.getOption("spark.master");
        if (!option.isDefined())   // use local over nothing
        JavaSparkContext ctx = new JavaSparkContext(sparkConf);

        String[] linesTXT = GETTYSBURG.split("\n"); // the gettysburg
address as lines of text
        JavaRDD<String> lines = ctx.parallelize(Arrays.asList(linesTXT));
        lines = lines.persist(StorageLevel.MEMORY_ONLY());

        // try for variant counts 10..32 by 2 - Each case is 4 times the
        for (int variantSize = 10; variantSize < 32; variantSize += 2) {
            long startMSec = System.currentTimeMillis();  // when did we
            // Drop all non-letters - make upper case split into groups of
size  variantSize
            JavaRDD<String> words = lines.flatMap(new

            // generate all variants with different case  should be  THE ->
            JavaRDD<String> variants = words.flatMap(new

            // same as Java word count - we actually expect all counts to
be 1
            JavaPairRDD<String, Integer> ones = variants.mapToPair(new
PairFunction<String, String, Integer>() {
                public Tuple2<String, Integer> call(String s) {
                    return new Tuple2<String, Integer>(s, 1);

            // same as Java word count - we actually expect all counts to
be 1
            JavaPairRDD<String, Integer> counts = ones.reduceByKey(new
Function2<Integer, Integer, Integer>() {
                public Integer call(Integer i1, Integer i2) {
                    return i1 + i2;
            List<Tuple2<String, Integer>> output =
            long endMSec = System.currentTimeMillis();   // when did we

            System.out.println("Variant Size " + variantSize + " Size " +
output.size() + " took " + (int) (endMSec - startMSec) / 1000);

View raw message