共计 5620 个字符，预计需要花费 15 分钟才能阅读完成。
本文次要探讨如何应用 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 能够如下设置：
Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp()
.setBootstrapServers("localhost:9092")
.setDataFormat("json")
.setTopic("iris");留神：Kafka 写入的数据只能为字符串，须要设置每条记录转化为字符串的形式，这里咱们应用 Json 格局。
咱们还须要结构一个获取流式数据的形式，最简略的形式是应用 CsvSourceStreamOp 组件，将 csv 数据（alink-release.oss-cn-beijing.aliyuncs.com）以流的形式读入。而后，再连贯 Kafka 写入组件，开始执行流式操作。残缺代码如下：
private static void writeKafka() throws Exception {
String URL = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv";
String SCHEMA_STR
= "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";
CsvSourceStreamOp data = new CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR);
Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp()
.setBootstrapServers("localhost:9092")
.setDataFormat("json")
.setTopic("iris");
data.link(sink);
StreamOperator.execute();}因为 CSV 文件中数据无限，当读取完最初一条时，流式工作会完结。
接下来，咱们能够应用 Alink 的 Kafka011SourceStreamOp 组件读取数据，并设置其消费者组 ID，读取模式为从头开始，具体代码如下：
private static void readKafka() throws Exception {Kafka011SourceStreamOp source = new Kafka011SourceStreamOp()
.setBootstrapServers("localhost:9092")
.setTopic("iris")
.setStartupMode("EARLIEST")
.setGroupId("alink_group");
source.print();
StreamOperator.execute();}执行打印后果如下，两头略去大部分数据：
message_key|message|topic|topic_partition|partition_offset
-----------|-------|-----|---------------|----------------
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.6}|iris|0|0
null|{"sepal_width":4.1,"petal_width":0.1,"sepal_length":5.2,"category":"Iris-setosa","petal_length":1.5}|iris|0|1
null|{"sepal_width":2.8,"petal_width":1.5,"sepal_length":6.5,"category":"Iris-versicolor","petal_length":4.6}|iris|0|2
null|{"sepal_width":3.0,"petal_width":1.8,"sepal_length":6.1,"category":"Iris-virginica","petal_length":4.9}|iris|0|3
null|{"sepal_width":2.9,"petal_width":1.8,"sepal_length":7.3,"category":"Iris-virginica","petal_length":6.3}|iris|0|4
......
null|{"sepal_width":2.2,"petal_width":1.0,"sepal_length":6.0,"category":"Iris-versicolor","petal_length":4.0}|iris|0|145
null|{"sepal_width":2.4,"petal_width":1.0,"sepal_length":5.5,"category":"Iris-versicolor","petal_length":3.7}|iris|0|146
null|{"sepal_width":3.1,"petal_width":0.2,"sepal_length":4.6,"category":"Iris-setosa","petal_length":1.5}|iris|0|147
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.9}|iris|0|148
null|{"sepal_width":2.9,"petal_width":1.4,"sepal_length":6.1,"category":"Iris-versicolor","petal_length":4.7}|iris|0|149能够看到间接从 Kafka 中获取的每条数据都是 Json 格局的字符串。
接下来，咱们须要对字符串外面的数据进行提取。举荐应用 JsonValueStreamOp，通过设置须要提取内容的 JsonPath，提取出各列数据。具体代码如下：
Kafka011SourceStreamOp source =
new Kafka011SourceStreamOp()
.setBootstrapServers("localhost:9092")
.setTopic("iris")
.setStartupMode("EARLIEST")
.setGroupId("alink_group");
StreamOperator data = source
.link(new JsonValueStreamOp()
.setSelectedCol("message")
.setReservedCols(new String[] {})
.setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"})
.setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width",
"$.category"})
);
System.out.print(data.getSchema());
data.print();
StreamOperator.execute();对于后果数据的 Schema 打印为：
root
|-- sepal_length: STRING
|-- sepal_width: STRING
|-- petal_length: STRING
|-- petal_width: STRING
|-- category: STRING能够看出 JsonValueStreamOp 提取进去的后果都是 string 类型的，具体数据打印后果如下，略去两头的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8|3.4|1.6|0.2|Iris-setosa
5.2|4.1|1.5|0.1|Iris-setosa
6.5|2.8|4.6|1.5|Iris-versicolor
6.1|3.0|4.9|1.8|Iris-virginica
7.3|2.9|6.3|1.8|Iris-virginica
......
5.2|2.7|3.9|1.4|Iris-versicolor
6.4|2.7|5.3|1.9|Iris-virginica
6.8|3.0|5.5|2.1|Iris-virginica
5.7|2.5|5.0|2.0|Iris-virginica
6.1|2.8|4.0|1.3|Iris-versicolor至此，咱们曾经可能拿到数据了，只是数据的类型有问题，须要进行转换。咱们能够应用 Flink SQL 的 cast 办法，在代码实现上，只需在连贯 JsonValueStreamOp 之 后，应用 select 办法（其参数为 SQL 语句），具体代码如下：
StreamOperator data = source
.link(new JsonValueStreamOp()
.setSelectedCol("message")
.setReservedCols(new String[] {})
.setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"})
.setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width",
"$.category"})
)
.select("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"
);执行新的代码，对于后果数据的 Schema 打印为：
root
|-- sepal_length: DOUBLE
|-- sepal_width: DOUBLE
|-- petal_length: DOUBLE
|-- petal_width: DOUBLE
|-- category: STRING每列数据都转化为相应的类型。具体数据打印后果如下，略去两头的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8000|3.4000|1.6000|0.2000|Iris-setosa
5.2000|4.1000|1.5000|0.1000|Iris-setosa
6.5000|2.8000|4.6000|1.5000|Iris-versicolor
6.1000|3.0000|4.9000|1.8000|Iris-virginica
7.3000|2.9000|6.3000|1.8000|Iris-virginica
......
5.2000|2.7000|3.9000|1.4000|Iris-versicolor
6.4000|2.7000|5.3000|1.9000|Iris-virginica
6.8000|3.0000|5.5000|2.1000|Iris-virginica
5.7000|2.5000|5.0000|2.0000|Iris-virginica
6.1000|2.8000|4.0000|1.3000|Iris-versicolor能够看出，配合应用的相干组件，能够残缺地从 Kafka 上读取、写入数据。前面，可通过 Alink 的各算法组件进行深刻计算。
以上。Alink 是基于 Flink 的机器学习算法平台，欢送拜访 Alink 的 GitHub 链接获取更多信息。也欢送退出 Alink 开源用户群进行交换～
Alink GitHub 链接：
https://github.com/alibaba/Alink
▼ 钉钉扫码退出 Alink 技术交换群 ▼
