查看和操作Kafka存储数据的快速方法？#isoBlue#isoBus

Question

以最快和最简单的方式，我希望二进制数据在Kafka消息日志可视为十六进制字符串值。

我有以下数据，其中包含二进制形式的CAN消息，我希望将其视为十六进制字符串

case-8010-wheat-ault-072018/
├── cleaner-offset-checkpoint
├── debug-0
│   ├── 00000000000000000000.index
│   ├── 00000000000000000000.log
│   ├── 00000000000000000000.timeindex
│   ├── 00000000000000006972.index
│   ├── 00000000000000006972.log
│   ├── 00000000000000006972.snapshot
│   ├── 00000000000000006972.timeindex
│   ├── 00000000000000079766.snapshot
│   └── leader-epoch-checkpoint
├── gps-0
│   ├── 00000000000000000000.index
│   ├── 00000000000000000000.log
│   ├── 00000000000000000000.timeindex
│   ├── 00000000000000003235.index
│   ├── 00000000000000003235.log
│   ├── 00000000000000003235.snapshot
│   ├── 00000000000000003235.timeindex
│   ├── 00000000000000029657.snapshot
│   └── leader-epoch-checkpoint
├── imp-0
│   ├── 00000000000000000000.index
│   ├── 00000000000000000000.log
│   ├── 00000000000000000000.timeindex
│   ├── 00000000000000004940.index
│   ├── 00000000000000004940.log
│   ├── 00000000000000004940.snapshot
│   ├── 00000000000000004940.timeindex
│   ├── 00000000000000915321.snapshot
│   └── leader-epoch-checkpoint

数据源使用kafka_2.11-0.11.0.1提供用于使用数据的文档，但可能是由于版本不匹配。tgz步骤失败。https://www.isoblue.org/docs/data/data/

试图直接查看数据，我有：

1.玩了kafka.tools.转储

./kafka-run-class.sh kafka.tools.DumpLogSegments --deep-iteration --print-data-log  --files ../../case-8010-wheat-ault-072018/imp-0/00000000000000004940.log | head -n 15

我试图操纵DumpLogSegments输出，但数据看起来不像预期的那样<代码>/Kafka跑步课。Kafka。工具。DumpLogSegments--深度迭代--打印数据日志--文件..//案例-8010-wheat-ault-072018/imp-0/0000000000000000 4940。log | awk-F'payload:“{print$2}”| awk-F'offset:“{print$1}”| od-A n-t x1 | head-n 10

2.我目前正在尝试基于以下内容编写文件源连接器：https://docs.confluent.io/current/connect/devguide.html#connector-范例

Answer 1

我无法让Kafka消费者与蟒蛇Kafka合作。isoBlue脚本对我不起作用。

1.编辑$KAFKA_HOME/config/server。属性<代码>日志。迪尔斯=

2.zookeeper服务器启动。sh$KAFKA_HOME/config/zookeeper。性质

3.Kafka主题。sh--列表--zookeeper本地主机：2181

4.Kafka服务器启动。sh$KAFKA_HOME/config/server。性质

5.使用蟒蛇3：

#!/usr/bin/env python3

from kafka import KafkaConsumer
import sys

try:

    bootstrap_servers = ['localhost:9092']
    topicName = 'tra'
    consumer = KafkaConsumer (topicName, group_id = 'can-test',bootstrap_servers = bootstrap_servers,auto_offset_reset='earliest');

    for message in consumer:
        wait = input("PRESS ENTER TO CONTINUE.")    
        print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))
except KeyboardInterrupt:
    sys.exit()

6.它真的帮助我观察服务器是否在运行，数据是否还在那里。

watch -n 5 ls -alF <KAFKA_DATA> | head
watch -n 5 netstat -a | grep 9092