标签: confluent-platform

我有以下格式的数据进入 Kafka。

{"WHS":[{"Character Set":"UTF-8","action":"finished","Update-Date-Time":"2020-04-11 09:00:02:25","Number":0,"Abbr":"","Name":"","Name2":"","Country-Code":"","Addr-1":"","Addr-2":"","Addr-3":"","Addr-4":"","City":"","State":""}]}

我希望它像这样转换。

{"Character Set":"UTF-8","action":"finished","Update-Date-Time":"2020-04-11 09:00:02:25","Number":0,"Abbr":"","Name":"","Name2":"","Country-Code":"","Addr-1":"","Addr-2":"","Addr-3":"","Addr-4":"","City":"","State":""}

我尝试使用 ksql 进行扁平化，但 ksql 还不支持数组。

我尝试kstream使用以下代码进行扁平化。

builder.stream(inputTopic).flatMapValues(Object -> Arrays.asList()).to(outputTopic);

但它没有产生任何输出。对此的任何帮助将不胜感激。

这是我的连接器配置：

curl -s -k -X POST http://***************:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "mysql-cdc-CUSTOMER_DETAILS-007",
  "config": {
    "tasks.max":"2",
    "poll.interval.ms":"500",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "dbnode",
    "database.port": "3306",
    "database.user": "**********",
    "database.password": "###########",
    "database.server.name": "dbnode",
    "database.whitelist": "device_details",
    "database.history.kafka.bootstrap.servers": "**********:9092",
    "database.history.kafka.topic": "schema-changes.device_details",
    "include.schema.changes":"true",
    "table.whitelist":"device_details.tb_customermst",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://************: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"
  }
}' | jq '.'

从ksql消费数据时，显示如下：

ksql> print 'Device_Details.device_details.tb_customermst' from beginning;
Format:AVRO
5/2/20 2:08:34 PM IST, Struct{customerid=10001}, {"before": null, "after": {"customerid": 10001, "firstname": "Klara", "lastname": "Djokic", "emailid": "klara.djokic007@iillii.org", "mobilenumber": "+1 …

我开始看到以下错误

[2020-06-12 20:09:01,324] ERROR [ReplicaManager broker=3] Error processing append operation on partition __consumer_offsets-10 (kafka.server.ReplicaManager)
org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the current ISR Set(3) is insufficient to satisfy the min.isr requirement of 2 for partition __consumer_offsets-10    

我的设置是拥有三个经纪人，并且所有经纪人都已上线。在出现此错误之前我做了几件事

我在所有代理中将 min.isr 配置为 2。我创建了一个复制因子为 3 的主题，并开始从 ack = 1 且两个代理关闭的生产者生成消息。我召集了所有经纪人并开始消费。

1. 如何解决此错误
2. 消费者也无法看到此消息（不知道为什么，该消息应该被视为“已提交”，因为当生产者运行时一个代理已启动）

几个事实

有趣的是，WRT 首选的领导者战略并未发生重新平衡

$ kafka-topics --zookeeper 127.0.0.1:2181 --topic stock-prices --describe                             
  Topic: stock-prices     PartitionCount: 3       ReplicationFactor: 3    Configs: min.insync.replicas=2
  Topic: stock-prices     Partition: 0    Leader: 1       Replicas: 1,3,2 Isr: 1,2,3            
  Topic: stock-prices     Partition: 1 …

使用现有的 Java 示例，我尝试使用 python-kafka 和 confluence_kafka 库编写一个与生成器等效的 python 版本。如何使用类似于下面 Java 中的信息在 python 中配置 sasl.jass.config？

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

...
Properties props = new Properties();
...
props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"<Kafka_Username>\" password=\"<Kafka_Password>\";");

Producer<String, String> producer = new KafkaProducer<>(props);

我有下面的 docker-compose.yml 文件

version: '2'

services:
zookeeper:
    image: confluentinc/cp-zookeeper:6.0.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
    - "2181:2181"
    environment:
    ZOOKEEPER_CLIENT_PORT: 2181
    ZOOKEEPER_TICK_TIME: 2000

kafka:
    image: confluentinc/cp-enterprise-kafka:6.0.0
    hostname: kafka
    container_name: kafka
    depends_on:
    - zookeeper
    ports:
    - "29092:29092"
    environment:
    KAFKA_BROKER_ID: 1
    KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

当我运行时docker-compose exec kafka bash，我收到 bash 提示符。

在 bash 提示符中，我已经kafka-console-consumer但无权访问kafka-avro-console-consumer?

我怎样才能访问kafka-avro-console-consumer？它不是在 $PATH 中而是在其他目录中吗？

我尝试使用find和which命令，但这些命令不存在于 docker 运行容器中

是否可以合并kafka中的记录并将输出发布到不同的流？

例如，有一个针对 kafka 主题的事件流，如下所示

{txnId:1,startTime:0900},{txnId:1,endTime:0905},{txnId:2,endTime:0912},{txnId:3,endTime:0930},{txnId:2,startTime:0912}, {txnId:3,开始时间:0925}......

我想通过 txnId 合并这些事件并创建合并的输出，如下所示

{txnId:1,startTime:0900,endTime:0905},{txnId:2,startTime:0910,endTime:0912},{txnId:3,startTime:0925,endTime:0930}

请注意，传入事件中不会维护顺序。因此，如果在开始时间事件之前收到 txn Id 的 endTime，那么我们需要等到收到该 txnId 的开始时间事件后再启动合并

我浏览了 Kafka Streams 示例附带的字数示例，但不清楚如何等待事件，然后在进行转换时合并。

任何想法都受到高度赞赏。

我最近在学习卡夫卡.Kafka和Confluent之间有什么关系？

我正在使用 Kafka 使多个微服务相互通信。服务是用 Python 编写的，我使用 Confluent 库来处理 Kafka。在某些时候，我知道某些主题刚刚“结束”，因此我可以自动清理它们。

由于 Confluent 库，有没有办法删除“主题”？我找不到任何关于此的文档...

谢谢

我使用 kafka-connect-jdbc-4.0.0.jar 和 postgresql-9.4-1206-jdbc41.jar

kafka connect的connector配置

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "mode": "timestamp",
  "timestamp.column.name": "updated_at",
  "topic.prefix": "streaming.data.v2",
  "connection.password": "password",
  "connection.user": "user",
  "schema.pattern": "test",
  "query": "select * from view_source",
  "connection.url": "jdbc:postgresql://host:5432/test?currentSchema=test"
}

我已经使用 jdbc 驱动程序配置了两个连接器一个源和另一个接收器，针对 postgresql 数据库（“PostgreSQL 9.6.9”）一切正常

我对连接器如何收集源数据有疑问，查看日志我看到执行查询之间存在 21 秒的时间差

11/1/2019 9:20:18[2019-01-11 08:20:18,985] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='null', query='select * from view_source', topicPrefix='streaming.data.v2', timestampColumn='updated_at', incrementingColumn='null'} (io.confluent.connect.jdbc.source.JdbcSourceTask)
11/1/2019 9:20:18[2019-01-11 08:20:18,985] DEBUG TimestampIncrementingTableQuerier{name='null', query='select * from view_source', topicPrefix='streaming.data.v2', timestampColumn='updated_at', incrementingColumn='null'} prepared SQL query: select * from view_source WHERE "updated_at" > …

我正在尝试使用 Golang 客户端测试生产者将消息写入 kafka 集群上的主题。这可以很好地写入本地集群上的主题，我只是从他们的github repo复制并粘贴了示例代码。

package main

import (
    "fmt"
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)


func main() {

    p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers":"localhost"})
    if err != nil {
        panic(err)
    }

    defer p.Close()

    // Delivery report handler for produced messages
    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
                } else {
                    fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
                }
            }
        }
    }()

    // Produce messages to topic …