使用Apache Spark为Kafka生成Avro主题

Question

我已经在本地安装了kafka（目前没有集群/模式注册），并尝试生成一个Avro主题，下面是与该主题相关的模式。

{
  "type" : "record",
  "name" : "Customer",
  "namespace" : "com.example.Customer",
  "doc" : "Class: Customer",
  "fields" : [ {
    "name" : "name",
    "type" : "string",
    "doc" : "Variable: Customer Name"
  }, {
    "name" : "salary",
    "type" : "double",
    "doc" : "Variable: Customer Salary"
  } ]
}

我想创建一个简单的SparkProducerApi来根据上述模式创建一些数据并将其发布到kafka。考虑创建转换为dataframe的示例数据，然后将其更改为avro然后发布。

val df = spark.createDataFrame(<<data>>)

然后，如下所示：

df.write
  .format("kafka")
  .option("kafka.bootstrap.servers","localhost:9092")
  .option("topic","customer_avro_topic")
  .save()
}

现在可以< code >手动将模式附加到此avro主题。

这可以通过使用Apache Spark API而不是使用Java/Kafka API来实现吗？这是用于批处理，而不是流式处理。

Answer 1

我认为这是不可能的，因为Spark中的Kafka生产者需要两列key和value，它们都必须是字节数组。

如果您从磁盘读取现有的Avro文件，那么您使用的Avro数据帧读取器可能会为名称和工资创建两列。因此，您需要一个操作来从包含整个Avro记录的其他列中构造一个＜code＞值

如果您想生成数据并且没有文件，那么您需要为Kafka消息键和作为字节数组的值构建一个Tuple2对象列表，然后您可以将它们并行化到RDD，然后将它们转换为Dataframe。但是在这一点上，只使用常规的Kafka Producer API要简单得多。

另外，如果您已经知道您的模式，请尝试在Kafka中生成测试数据的方法中提到的项目