Description

kafka 011 sink.

Parameters

Name Description Type Required？ Default Value
bootstrapServers kafka集群地址 String ✓
topic topic String ✓
dataFormat data format String ✓
fieldDelimiter Field delimiter String “,”

Script Example

  1. URL = "http://alink-dataset.cn-hangzhou.oss.aliyun-inc.com/csv/iris.csv";
  2. SCHEMA_STR = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";
  3. data = CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR)
  5. sink = Kafka011SinkStreamOp() \
  6.     .setBootstrapServers("localhost:9092").setDataFormat("json") \
  7.     .setTopic("iris")
  8. sink.linkFrom(data)
  9. StreamOperator.execute()