Преобразование задания Hadoop в Spark

Question

Я пытаюсь перенести следующую работу Hadoop на Spark.

public class TextToSequenceJob { 

    public static void main(String[] args) throws IOException, 
      InterruptedException, ClassNotFoundException { 

     Job job = Job.getInstance(new Configuration()); 
     job.setJarByClass(Mapper.class); 

     job.setOutputKeyClass(LongWritable.class); 
     job.setOutputValueClass(Text.class); 

     job.setMapperClass(Mapper.class); 

     job.setInputFormatClass(TextInputFormat.class); 
     job.setOutputFormatClass(SequenceFileOutputFormat.class); 

     FileInputFormat.setInputPaths(job, new Path("input")); 
     FileOutputFormat.setOutputPath(job, new Path("output")); 

     job.submit(); 
    } 
} 

Это мой Спарк решение до сих пор:

public final class text2seq2 { 

    public static void main(String[] args) throws Exception { 

     SparkConf sparkConf = new SparkConf(); 

     sparkConf.setMaster("local").setAppName("txt2seq").set("spark.executor.memory", "1g"); 
     sparkConf.set("spark.executor.memory", "1g"); 
     JavaSparkContext ctx = new JavaSparkContext(sparkConf); 
     JavaPairRDD<String, String> infile = ctx.wholeTextFiles("input"); 
     infile.saveAsHadoopFile("output", LongWritable.class, String.class, SequenceFileOutputFormat.class); 
     ctx.stop(); 
    } 
} 

Но я получаю эту ошибку:

java.io.IOException: Could not find a serializer for the Value class: 'java.lang.String'. Please ensure that the configuration 'io.serializations' is properly configured, if you're usingcustom serialization.
at org.apache.hadoop.io.SequenceFile$Writer.init(SequenceFile.java:1187)

Кто-нибудь знает, что это значит?

Answer 1

Я понял это как показано ниже. Надеюсь, что это будет полезно для некоторых людей.

public final class text2seq { 

    public static void main(String[] args) throws Exception { 

     SparkConf sparkConf = new SparkConf(); 

     sparkConf.setMaster("local").setAppName("txt2seq").set("spark.executor.memory", "1g"); 
     sparkConf.set("spark.executor.memory", "1g"); 
     JavaSparkContext ctx = new JavaSparkContext(sparkConf); 
     JavaRDD<String> infile = ctx.textFile("input"); 
     JavaPairRDD<LongWritable, Text> pair = infile.mapToPair(new PairFunction<String, LongWritable, Text>() { 
      /** 
      * 
      */ 
      private static final long serialVersionUID = 1L; 

      @Override 
      public Tuple2<LongWritable, Text> call(String s) { 
       return new Tuple2<LongWritable, Text>(new LongWritable(), new Text(s)); 
      } 
     }); 
     pair.saveAsHadoopFile("output", LongWritable.class, Text.class, 
       SequenceFileOutputFormat.class); 
     ctx.stop(); 
    } 
}