I have a set of files that span multiple lines, so we are using Unicode character \\u0003 to denote the end of a record. However, my Spark code is ignoring the instruction to use the alternate delimiter, so each line is being returned as an entry.
Here is the relevant snippets of code:
SparkConf conf = new SparkConf();
conf.setMaster("local[2]");
conf.setAppName("MyApp");
Configuration config = new Configuration();
config.set("textinputformat.record.delimiter", "\\u0003");
conf.set("textinputformat.record.delimiter", "\\u0003");
final int batchSize = Integer.parseInt(cli.getOptionValue(BATCH_SIZE));
final JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(STREAM_DURATION_IN_SECS));
JavaDStream<String> inputStream = jsc.textFileStream(cli.getOptionValue(INPUT_DIRECTORY));
JavaDStream<Record> records = inputStream.flatMap(new CacheFileStreamFlatMap());
//in the above function I have a Log message where I can see the records coming out line by line
boolean mapSideCombine = true;
JavaPairDStream<String, Record> iniRecords = records.flatMapToPair(new RecordToIniRecord());
JavaPairDStream<String, List<Record>> combined = iniRecords.combineByKey(new CreateListOfRecords(),
// Create combiners
new AddRecordToList(),
// merge Values
new MergeListsOfRecords(), new HashPartitioner(batchSize), mapSideCombine);
combined.foreach(new WriteRecordsToIniTable());
jsc.start(); // Start the computation
jsc.awaitTermination(); // Wait for the computation to terminate
return 0;
Every entry that I can find in StackOverflow points at this as being the right way to accomplish this task, but I'm not seeing it work.
Thanks!
