如何在Spark中将数据序列化为AVRO模式(用Java)？

Question

我定义了一个AVRO模式，并使用AVRO工具为这些模式生成了一些类。现在，我想将数据序列化到磁盘。我找到了一些关于scala的答案，但不适用于Java。类＜code＞文章

下面是我如何尝试这样做的代码的简化版本：

JavaPairRDD<String, String> filesRDD = context.wholeTextFiles(inputDataPath);
JavaRDD<Article> processingFiles = filesRDD.map(fileNameContent -> {
    // The name of the file
    String fileName = fileNameContent._1();
    // The content of the file
    String fileContent = fileNameContent._2();

    // An object from my avro schema
    Article a = new Article(fileContent);

    Processing processing = new Processing();
    // .... some processing of the content here ... //

    processing.serializeArticleToDisk(avroFileName);

    return a;
});

其中序列化ArticleToDisk（avroFileName）定义如下：

public void serializeArticleToDisk(String filename) throws IOException{
    // Serialize article to disk
    DatumWriter<Article> articleDatumWriter = new SpecificDatumWriter<Article>(Article.class);
    DataFileWriter<Article> dataFileWriter = new DataFileWriter<Article>(articleDatumWriter);
    dataFileWriter.create(this.article.getSchema(), new File(filename));
    dataFileWriter.append(this.article);
    dataFileWriter.close();
}

其中文章是我的avro模式。

现在，映射器向我抛出错误：

java.io.FileNotFoundException: hdfs:/...path.../avroFileName.avro (No such file or directory)   
at java.io.FileOutputStream.open0(Native Method)    
at java.io.FileOutputStream.open(FileOutputStream.java:270)     
at java.io.FileOutputStream.<init>(FileOutputStream.java:213)   
at java.io.FileOutputStream.<init>(FileOutputStream.java:162)   
at org.apache.avro.file.SyncableFileOutputStream.<init>(SyncableFileOutputStream.java:60)   
at org.apache.avro.file.DataFileWriter.create(DataFileWriter.java:129)
at org.apache.avro.file.DataFileWriter.create(DataFileWriter.java:129)
at sentences.ProcessXML.serializeArticleToDisk(ProcessXML.java:207)     
. . . rest of the stacktrace ... 

虽然文件路径是正确的。

之后我使用了 collect（） 方法，因此 map 函数中的其他所有内容都可以正常工作（序列化部分除外）。

我对Spark很陌生，所以我不确定这是否可能是微不足道的事情。我怀疑我需要使用一些写入函数，而不是在映射器中进行写入（尽管不确定这是否属实）。有什么想法如何解决这个问题？

编辑：

我展示的错误堆栈跟踪的最后一行实际上在这一部分:

dataFileWriter.create（this.article.getSchema（）， new File（filename））;

这是抛出实际错误的部分。我假设< code>dataFileWriter需要用其他东西来替换。有什么想法吗？

Answer 1

看来你使用Spark的方式不对。

< code>Map是一个转换函数。仅调用< code>map不会调用< code>RDD的计算。您必须调用像< code>forEach()或< code>collect()这样的操作。

还要注意，提供给< code > map lambda将在驱动程序中序列化，并传输到群集中的某个< code >节点。

添加

尝试使用 Spark SQL 和 Spark-Avro 以 Avro 格式保存 Spark DataFrame：

// Load a text file and convert each line to a JavaBean.
JavaRDD<Person> people = sc.textFile("/examples/people.txt")
    .map(Person::parse);

// Apply a schema to an RDD
DataFrame peopleDF = sqlContext.createDataFrame(people, Person.class);
peopleDF.write()
    .format("com.databricks.spark.avro")
    .save("/output");

Answer 2

此解决方案不使用数据帧并且不会抛出任何错误：

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.io.NullWritable;
import org.apache.avro.mapred.AvroKey;
import org.apache.spark.api.java.JavaPairRDD;
import scala.Tuple2;

   .  .  .  .  .

// Serializing to AVRO
JavaPairRDD<AvroKey<Article>, NullWritable> javaPairRDD = processingFiles.mapToPair(r -> {    
    return new Tuple2<AvroKey<Article>, NullWritable>(new AvroKey<Article>(r), NullWritable.get());
});
Job job = AvroUtils.getJobOutputKeyAvroSchema(Article.getClassSchema());
javaPairRDD.saveAsNewAPIHadoopFile(outputDataPath, AvroKey.class, NullWritable.class, AvroKeyOutputFormat.class, 
        job.getConfiguration());

其中AvroUtils.getJobOutputKeyAvroSchema是：

public static Job getJobOutputKeyAvroSchema(Schema avroSchema) {
    Job job;

    try {
        job = new Job();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }

    AvroJob.setOutputKeySchema(job, avroSchema);
    return job;
}

Spark Avro的类似内容可以在这里找到 -