在 HDFS 中写入 avro 文件 - 存在

Question

目前我正在学习 Spark Streaming 和 avro，所以我的第一个示例是，读取 Spark RDD 并构建通用记录，创建 avro 文件，这个文件我应该在 HDFS 中写入。现在我可以打开 avro 文件并且我确实附加到 HDFS 文件存在吗？

这段代码编写了一个 avro 文件，但是当我尝试添加或追加时，它失败了。我为此使用 java 8

public static void saveAvro(GenericRecord record, Schema schema) throws IOException {

        DatumWriter<GenericRecord> bdPersonDatumWriter = new GenericDatumWriter<>(schema);
        DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(bdPersonDatumWriter);

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create("hdfs://sandbox-hdp.hortonworks.com:8020/tmp/poc/ResultHDFSTest.avro"),
                conf);
        Path F = new Path("hdfs://sandbox-hdp.hortonworks.com:8020/tmp/poc/ResultHDFSTest.avro");
        fs.setReplication(F, (short) 1);

        if (!fs.exists(F)) {
            System.out.println("File not exists.. creating....");
            OutputStream out = fs.create(F, (short) 1);
            System.out.println("OutputStream create.");
            dataFileWriter.create(schema, out);
            System.out.println("dataFileWriter create.");
            dataFileWriter.append(record);
            System.out.println("dataFileWriter append OK {0} .");

        } else {
            //Here fail, not open file.. avro stored in HDFS
            System.out.println("File exists....");
           // I want to add information to an existing avro file.
            dataFileWriter.append(record);
            System.out.println("dataFileWriter append OK {1} .");
        }
        dataFileWriter.close();
        System.out.println("dataFileWriter closed.");

    }
    

追加存在文件 avro HDFS 的堆栈跟踪：

线程“主”org.apache.avro.AvroRuntimeException 中的异常：不是 打开 在 org.apache.avro.file.DataFileWriter.assertOpen(DataFileWriter.java:88) 在 org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:311) 在 com.test.avro.App.saveAvro(App.java:83) 在 com.test.avro.App.main(App.java:55)

DataFileWriter appendTo 方法只接受 File java.nio。我正在尝试做的事情是正确的还是有其他方法？

编辑 1. 我想向现有文件添加信息。

第一个代码 sn-p 显示了您尝试创建 avro 文件的实现。这是我的火花流的框架代码：

JavaStreamingContext jssc = sparkConfigurationBuilder
                .buildJSC(sparkConfigurationBuilder.buildSparkConfiguration());
    
    jssc.sparkContext().checkpointFile("c:\\tmp");
    Map<String, Object> kafkaParams = sparkDriverUtils.getKafkaProperties();        
    Collection<String> topics = Arrays.asList(sparkDriverUtils.getTopics().trim().split(","));// 1 o more topics        
    LOGGER.warn("Lista de Topics: " + topics.toString());
    

...

JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
//This DSTream resulto to avro..
JavaDStream<Transactions> transactionsDS = transactions.map(f-> {
            Transactions txn = jsonMapperUtil.rowToTransaction(f);
            LOGGER.warn("Retornar  : JavaDStream<Transactions>");
            return  txn;
        });

现在 transactionsDS 结果我想在 HDFS 中保存为 avro 文件。我有一个问题，我可以获取 JavaStreamingContext 来为数据集创建 SparkSession，还是应该更改订阅 kafka 代理的方式？

问候。

Answer 1

DataFileWriter appendTo 方法只接受一个 File java.nio

正确。 Avro 没有连接到 HDFS 路径。

为了“附加到 HDFS 文件”，您需要将它们下载到本地，然后覆盖它们的全部内容

---

除此之外，您还提到了 Spark Streaming，但显示的代码中没有任何部分实际使用 Spark API 调用