所以我设置了Confluent Kafka JDBC连接器。首先,我启动一个模式注册表,如
./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties
这是schema-registery.properties文件
listeners=http://0.0.0.0:8081
kafkastore.connection.url=zookeeperhost:2181
kafkastore.bootstrap.servers=PLAINTEXT://kafkahost:9092
kafkastore.topic=_schemas
debug=false
接下来我启动一个像这样的独立连接器
./bin/connect-standalone ./etc/schema-registry/connect-avro-standalone.properties ./jdbc-source.properties
connect-avro-standalone.properties是
bootstrap.servers=kafkahost:9092
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
plugin.path=share/java
jdbc-source.properties是
name=jdbc_source_oracle
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=host)(PORT=port))(CONNECT_DATA=(SERVER=dedicated)(SID=server)))
connection.user=xxx
connection.password=xxx
table.whitelist=table1, table2
mode=bulk
topic.prefix=my_topic
query=select * from table1 t1 join table1 t2 on t2.id = t1.id where t2.entereddate >='19-FEB-2019' and t2.entereddate <= '23-FEB-2019'
我使用的查询仅用于测试目的,我想使用的实际查询将实现递增模式,并且不包含where子句。
现在,它设法将数据发布到主题中,但有一些奇怪的东西在继续。首先,ID以不可读的格式保存。只是空方块。其次,数据库中填充的某些字段在主题中保存为null。第三,每当我尝试在JDBC源文件中更改查询中的日期时,都不会发生任何事情。它仍然包含我第一次运行时发布的相同消息,因为kafka主题中的任何内容都没有更新,我更改了多少次查询。
谁能帮我?
编辑
我想要做的是通过pyspark代码消耗数据。这是关于我如何做的代码
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName("data streaming app")\
.getOrCreate()
data_raw = spark.readStream\
.format("kafka")\
.option("kafka.bootstrap.servers", "kafkahost:9092")\
.option("subscribe", "my_topic")\
.load()
query = data_raw.writeStream\
.outputMode("append")\
.format("console")\
.option("truncate", "false")\
.trigger(processingTime="5 seconds")\
.start()\
.awaitTermination()
我还使用此命令使用kafka-avro-console-consumer来使用数据
./bin/kafka-avro-console-consumer \
--bootstrap-server kafkahost:9092 \
--property print.key=true \
--from-beginning \
--topic my_topic
这两个都给我带来了奇怪的结果
这就是使用avro控制台给我的东西
阻止某些字段和文本,因为它可能包含公司敏感信息。
如果你从Spark消费Avro,你需要使用correct deserializer。
你从控制台看到你的Avro数据中的字节,然后是小数/数字,as detailed here的处理。
您可以阅读有关Kafka Connect和Avro(包括JSON)here的序列化替代品的更多信息。