从 Kafka 主题读取消息并将其转储到 HDFS

Question

我正在尝试使用来自 Kafka 主题的数据，将其加载到数据集中，然后在加载到 Hdfs 之前执行过滤。

我能够从 kafka 主题中消费，将其加载到数据集中并保存为 HDFS 中的镶木地板文件，但无法执行过滤条件。你能分享一下在保存到hdfs之前执行过滤的方法吗？ 我正在使用 Java 和 Spark 从 kafka 主题中消费。 我的部分代码是这样的：

DataframeDeserializer dataframe = new DataframeDeserializer(dataset);

 ds = dataframe.fromConfluentAvro("value", <your schema path>, <yourmap>, RETAIN_SELECTED_COLUMN_ONLY$.MODULE$);

StreamingQuery query = ds.coalesce(10)
                .writeStream()
                .format("parquet")
                .option("path", path.toString())
                .option("checkpointLocation", "<your path>")
                .trigger(Trigger.Once())
                .start();

Answer 1

我强烈推荐Kafka Connect，而不是重新发明轮子。 您只需要 HDFS Sink 连接器，它将数据从 Kafka 主题复制到 HDFS。

• 对于 HDFS 2.x 文件，您可以使用 HDFS 2 Sink Connector
• 对于 HDFS 3.x 文件，使用 HDFS 3 Sink Connector

Answer 2

在coalesce之前写过滤逻辑，即ds.filter().coalesce()

DataframeDeserializer dataframe = new DataframeDeserializer(dataset);

 ds = dataframe.fromConfluentAvro("value", <your schema path>, <yourmap>, RETAIN_SELECTED_COLUMN_ONLY$.MODULE$);

StreamingQuery query = 
                ds
                .filter(...) // Write your filter condition here
                .coalesce(10)
                .writeStream()
                .format("parquet")
                .option("path", path.toString())
                .option("checkpointLocation", "<your path>")
                .trigger(Trigger.Once())
                .start();