PROSAGA码农传奇-Kafka-Spark Dataframe到Kafka

Spark Dataframe到Kafka

作者: 张三岁
发布时间: 2024-03-21 10:08:28 (28天前)
转自：

<div class =“post-text”itemprop =“text”>
  
    以下是在流媒体中生成kafka的示例，但批处理版本几乎相同
  
  
    从源码流到kafka：
  
   <pre>
 <code>
 val ds = df
 .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
 .writeStream
 .format("kafka")
 .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
 .option("topic", "topic1")
 .start()

</code>
 </pre>
  
    编写静态数据帧（不是从源流式传输）到kafka
  
   <pre>
 <code>
 df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
 .write
 .format("kafka")
 .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
 .option("topic", "topic1")
 .save()

</code>
 </pre>
  
    请记住这一点
  
  <OL>
    <LI>
      每一行都是一条消息。
    </LI>
    <LI>
      数据帧必须是流数据帧。如果您有静态数据帧，则使用静态版本。
    </LI>
  </醇>
  
    看看基本文档：
    <a href="https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html" rel="nofollow noreferrer">
      https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html
    </A>
  
  
    听起来你有一个静态数据帧，它不是来自源的流。
  
</DIV>