小编Rob*_*att的帖子

在我目前的用例中,我使用Spark核心从MS SQL Server读取数据并对数据进行一些处理并每隔1分钟将其发送到Kafka,我使用Spark和Phoenix来维护HBase表中的CDC信息.

但是这种设计存在一些问题,例如,如果MS SQL记录激增,Spark处理比批处理间隔花费更多时间,并且spark最终会向Kafka发送重复记录.

作为替代方案,我正在考虑使用Kafka Connect从MS SQL读取消息并将记录发送到Kafka主题并在Kafka中维护MS SQL CDC.Spark Streaming将从Kafka主题中读取记录,并将记录和存储处理到HBase并发送到其他Kafka主题.

我有几个问题要实现这个架构:

1. 我可以使用开源Kafka连接器和Apache Kafka 0.9版本来实现这种架构.

2. 如果是的话,请你推荐一个GitHub项目,它可以为我提供这样的连接器,我可以使用SQL查询CDC MS SQL表SELECT * FROM SOMETHING WHERE COLUMN > ${lastExtractUnixTime}),并将记录存储到Kafka主题中.

3. Kafka connect是否支持Kerberos Kafka设置.

我们正在尝试使用 Confluent JDBC Sink 连接器将主题中的值写回 postgres 数据库。

connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
connection.password=xxx
tasks.max=1
topics=topic_name
auto.evolve=true
connection.user=confluent_rw
auto.create=true
connection.url=jdbc:postgresql://x.x.x.x:5432/Datawarehouse
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081

我们可以使用以下命令读取控制台中的值：

kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic topic_name

模式存在并且值被正确反序列化，kafka-avro-console-consumer因为它没有给出错误但连接器给出了这些错误：

  {
  "name": "datawarehouse_sink",
  "connector": {
    "state": "RUNNING",
    "worker_id": "x.x.x.x:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "FAILED",
      "worker_id": "x.x.x.x:8083",
      "trace": "org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:511)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:491)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:322)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.DataException: f_machinestate_sink\n\tat io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:103)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$0(WorkerSinkTask.java:511)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)\n\t... 13 more\nCaused by: org.apache.kafka.common.errors.SerializationException: Error …

我正在运行 kafka kubenetes helm 部署，但是我不确定如何安装自定义插件。

在本地版本的 kafka 上运行自定义插件时，我将卷安装/myplugin到 Docker 映像，然后设置插件路径环境变量。

我不确定如何将此工作流程应用于 helm Charts/kubernetes 部署，主要是如何将插件安装到 Kafka Connect pod，以便可以在 default 中找到它plugin.path=/usr/share/java。

我刚刚开始学习 Kafka，我正在尝试构建一个原型来拥有一个 REST API 生产者并将数据发送给 Kafka 消费者。我查阅了大量文档来找出一些特定的程序。

我无法理解是否有一个连接器可以像为 Apache Kafka 提供的文件连接器或 JDBC 连接器一样使用。我应该为此编写一个自定义连接器吗？

我很困惑从哪里开始。我特别寻找一些关于如何完成这项工作的结构化文档或想法。

我是stackoverflow的新手,所以让我知道如果我在这里发布这个问题我有什么不对.

我已经尝试找到答案,但无法在网站上找到KSQL JOIN相关问题,所以我发布了这个.我已经尝试了不同的方法来运行此查询,但我一直得到空指针异常,所以在此处发布.

我有两个kafka avro主题交易和费用,但数据有很多空白,以清楚我已创建以下主题和表与修剪数据.DEAL_STREAM和EXPENSE_TABLE

ksql> describe EXPENSE_TABLE;

结果:

Field      | Type
ROWTIME    | BIGINT (system)
ROWKEY     | VARCHAR(STRING) (system)
KSQL_COL_0 | VARCHAR(STRING)
KSQL_COL_1 | VARCHAR(STRING)
KSQL_COL_2 | VARCHAR(STRING)

和

ksql> describe deal_stream;

结果:

Field      | Type
ROWTIME    | BIGINT (system)
ROWKEY     | VARCHAR(STRING) (system)
KSQL_COL_0 | VARCHAR(STRING)
KSQL_COL_1 | VARCHAR(STRING)
KSQL_COL_2 | VARCHAR(STRING)

当我执行以下Query时,它给我空指针异常.我尝试了以下查询.

1:

ksql> CREATE STREAM deal_expense_new AS SELECT td.KSQL_COL_0 , te.KSQL_COL_1 FROM deal_stream td LEFT JOIN expense_table te ON td.KSQL_COL_0 = te.KSQL_COL_0;

2: …

我已经阅读了他们的博客并理解了他们的例子。 https://www.confluent.io/blog/apache-kafka-to-amazon-s3-exactly-once/

但我正在努力解决我所拥有的这种情况。我目前的配置是：

"flush.size": "50",
"rotate.interval.ms": "-1",
"rotate.schedule.interval.ms": "300000",
"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
"partition.duration.ms": "3600000",
"path.format": "YYYY/MM/dd/HH",
"timestamp.extractor": "Wallclock"

根据我对配置的了解。连接器将50在300000ms（5 分钟）后提交记录文件或文件，以先到者为准。如果连接器将文件上传到 s3 但未能提交到 Kafka，由于我设置了轮换计划间隔，Kafka 如何重新上传将覆盖 s3 文件的相同记录？这不会导致 s3 重复吗？