Kafka导出 (KafkaSinkStreamOp)

Java 类名：com.alibaba.alink.operator.stream.sink.KafkaSinkStreamOp
Python 类名：KafkaSinkStreamOp

功能介绍

写Kafka Plugin版。Kafka是由Apache软件基金会开发的一个开源流处理平台。详情
请参阅：https://kafka.apache.org/

参数说明

| 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| bootstrapServers | bootstrapServers | bootstrapServers | String | ✓ | | |

| dataFormat | 数据格式 | 数据格式。json,csv | String | ✓ | “JSON”, “CSV” | |

| topic | topic名称 | topic名称 | String | ✓ | | |

| fieldDelimiter | 字段分隔符 | 字段分隔符 | String | | | “,” |

| properties | 用户自定义Kafka参数 | 用户自定义Kafka参数,形如: “prop1= val1, prop2 = val2” | String | | | null |

代码示例

以下代码仅用于示意，可能需要修改部分代码或者配置环境后才能正常运行！

Python 代码

URL = "https://alink-test-data.oss-cn-hangzhou.aliyuncs.com/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 = KafkaSinkStreamOp() \
    .setBootstrapServers("localhost:9092").setDataFormat("json") \
    .setTopic("iris")
sink.linkFrom(data)
StreamOperator.execute()

Java 代码

import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.sink.KafkaSinkStreamOp;
import com.alibaba.alink.operator.stream.source.CsvSourceStreamOp;
import org.junit.Test;
public class KafkaSinkStreamOpTest {
    @Test
    public void testKafkaSinkStreamOp() throws Exception {
        String URL = "https://alink-test-data.oss-cn-hangzhou.aliyuncs.com/iris.csv";
        String SCHEMA_STR
            = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";
        StreamOperator <?> data = new CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR);
        StreamOperator <?> sink = new KafkaSinkStreamOp()
            .setBootstrapServers("localhost:9092").setDataFormat("json")
            .setTopic("iris");
        sink.linkFrom(data);
        StreamOperator.execute();
    }
}