关于机器学习:如何使用-Alink-连接-Kafka-数据源Python-版本

本文次要探讨如何应用 Alink 的 Kafka 连贯组件（Kafka011SourceStreamOp 和 Kafka011SinkStreamOp）读取写入数据。如何你须要一个本地的 Kafka 数据源进行试验，能够参考我另外一篇文章，具体介绍了搭建 Kafka 及建设 Topic 的过程。

• 在 MacOS 上搭建 Kafka
  https://zhuanlan.zhihu.com/p/…
• 在 Windows 上搭建 Kafka
  https://zhuanlan.zhihu.com/p/…

首先，咱们演示如何将流式数据写入 Kafka。

假如曾经有一个 Kafka 的数据源（譬如：本地 Kafka 数据源，端口为 9092），并且 Kafka 中曾经有一个 topic，名称为 iris，则 Kafka 写入组件 Kafka011SinkStreamOp 能够如下设置：

sink = Kafka011SinkStreamOp()\
.setBootstrapServers("localhost:9092")\
.setDataFormat("json")\
.setTopic("iris")

留神：Kafka 写入的数据只能为字符串，须要设置每条记录转化为字符串的形式，这里咱们应用 Json 格局。

咱们还须要结构一个获取流式数据的形式，最简略的形式是应用 CsvSourceStreamOp 组件，将 csv 数据（alink-release.oss-cn-beijing.aliyuncs.com）以流的形式读入。而后，再连贯 Kafka 写入组件，开始执行流式操作。残缺代码如下：

URL = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv"
SCHEMA_STR = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string"
data = CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR)

sink = Kafka011SinkStreamOp()\
.setBootstrapServers("localhost:9092")\
.setDataFormat("json")\
.setTopic("iris")

data.link(sink)

StreamOperator.execute()

因为 CSV 文件中数据无限，当读取完最初一条时，流式工作会完结。

接下来，咱们能够应用 Alink 的 Kafka011SourceStreamOp 组件读取数据，并设置其消费者组 ID，读取模式为从头开始，具体代码如下：

source = Kafka011SourceStreamOp()\
.setBootstrapServers("localhost:9092")\
.setTopic("iris")\
.setStartupMode("EARLIEST")\
.setGroupId("alink_group")

source.print(key='kafka_iris', refreshInterval=1, maxLimit=500)

StreamOperator.execute()

执行显示后果如下，PyAlink 对于流式数据的打印，是按肯定的工夫距离，每次显示若干条（默认是 100）。

message 列显示的每条数据的信息，能够看到间接从 Kafka 中获取的每条数据都是 Json 格局的字符串。

咱们在看到流式数据的同时，也留神到，连贯 Kafka 的流式工作始终没有退出，须要咱们点击菜单栏上的“中断”按钮，强行结束任务。

中断工作后，显示：

留神：之后，还须要点击 Restart 按钮，重启整个引擎，能力持续前面的操作。

接下来，咱们须要对字符串外面的数据进行提取。举荐应用 JsonValueStreamOp，通过设置须要提取内容的 JsonPath，提取出各列数据。具体代码如下：

source = Kafka011SourceStreamOp()\
.setBootstrapServers("localhost:9092")\
.setTopic("iris")\
.setStartupMode("EARLIEST")\
.setGroupId("alink_group")

data = source.link(JsonValueStreamOp()\
    .setSelectedCol("message")
    .setReservedCols([])
    .setOutputCols(["sepal_length", "sepal_width", "petal_length", "petal_width", "category"])
    .setJsonPath(["$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"])
)

data.getColTypes()

对于后果数据的 Schema 打印为：

[object, object, object, object, object]

能够看出 JsonValueStreamOp 提取进去的后果都是 object 类型的。咱们能够应用 Flink SQL 的 cast 办法，在代码实现上，只需在连贯 JsonValueStreamOp 之后，再连贯 SelectStreamOp 并设置其 SQL 语句参数，具体代码如下：

source = Kafka011SourceStreamOp()\
.setBootstrapServers("localhost:9092")\
.setTopic("iris")\
.setStartupMode("EARLIEST")\
.setGroupId("alink_group")

data = source.link(JsonValueStreamOp()\
    .setSelectedCol("message")
    .setReservedCols([])
    .setOutputCols(["sepal_length", "sepal_width", "petal_length", "petal_width", "category"])
    .setJsonPath(["$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"])
).link(SelectStreamOp()\
    .setClause("CAST(sepal_length AS DOUBLE) AS sepal_length,"\
               + "CAST(sepal_width AS DOUBLE) AS sepal_width,"\
               + "CAST(petal_length AS DOUBLE) AS petal_length,"\
               + "CAST(petal_width AS DOUBLE) AS petal_width, category")
)

data.getColTypes()

执行后果为：

[numpy.float64, numpy.float64, numpy.float64, numpy.float64, object]

即，每列数据都转化为相应的类型。

上面，咱们再对 data 进行打印，并开始执行流式工作，具体代码如下：

data.print()
    
StreamOperator.execute()

流式数据打印显示如下图所示：

能够看出，配合应用 Alink 的相干组件，能够残缺地从 Kafka 上读取、写入数据。前面，可通过 Alink 的各算法组件进行深刻计算。

以上。Alink 是基于 Flink 的机器学习算法平台，欢送拜访 Alink 的 GitHub 链接获取更多信息。也欢送退出 Alink 开源用户群进行交换～

Alink GitHub 链接：
https://github.com/alibaba/Alink

▼ 钉钉扫码退出 Alink 技术交换群 ▼