怎样安装Kafka？

Posted 2023-04-06 金木讲编程

1、Kafka是Java开发的应用程序，可以运行在Windows、 MacOS和 Linux等多 种操作系统上。最常见的是将Kafka安装在Linux系统上。
2、在安装Kafka之前，需要先安装Java环境，虽然运行 Zookeeper 和 Kafka 只需要 Java运行时版本，但也可以安装完整的 JDK，以备不时之需。
3、安装Zookeeper
Kafka使用 Zookeeper保存集群的元数据信息和消费者信息。 Kafka发行版自带了
Zookeeper，可以直接从脚本启动，不过安装一个完整版的 Zookeeper也并不费劲。

4. 安装 Kafka Broker
配置好 Java 和 Zookeeper 之后 ，接下来就可以安装 Kafka 了。可以从 http://kafka.apache. org/downloads.html 下载最新版本的 Kafka。安装命令如下：

我怎样才能从 kafka 主题接收数据到我的 Streaming Structured DataFrame？

【中文标题】我怎样才能从 kafka 主题接收数据到我的 Streaming Structured DataFrame？

【英文标题】：How could i get receive the data from kafka topic to my Streaming Structured DataFrame?

【发布时间】：2021-12-22 01:23:31

【问题描述】：

我知道如何使用我的 Kafka 主题中的数据，但我无法在正确的列中获取正确的数据。

我在 value 列中收到所有数据，格式如下：

"timestamp":"2021-11-09T11:03:48.955+01:00","time":"1","duration":"0","SourceComputer":"C1707","SourcePort":"N1","DestinationComputer":"C925","start/end":" "

有些字段是空的“”，而其他字段里面有一些数据（例如：“C1707”）。 我在想我可以使用这个功能：

DataFrame=DataFrame.withColumn(ColumnName[i],split(DataFrame["value"],',').getItem(i))

但我无法获得确切列中的确切数据。

+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+
|               value|           timestamp|time|duration|SourceComputer|SourcePort|DestinationComputer|start/end|
+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+
|"timestamp":"202...|"timestamp":"202...|null|    null|          null|      null|               null|     null|
+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+

知道如何以正确的方式接收数据吗？

谢谢！

【问题讨论】：

请出示完整的 Spark 代码。其他，请阅读databricks.com/blog/2017/04/26/…

【参考方案1】：

使用 from_json 方法，您可以将 Spark DataFrame 列上的 JSON 字符串转换为结构类型。 然后你可以将你的结构类型转换为所需的数据帧

import org.apache.spark.sql.functions.from_json
val schema = new StructType()
    .add("col1", StringType, true)
    .add("col2", StringType, true)
    .add("col3", StringType, true)
val df4=df.withColumn("value",from_json(col("value"),schema))
val df5=df4.select(col("value.*"))

请参考this link，这里都说明了。