Kafka Reader

Apache Kafka是一个快速、可扩展、高吞吐和可容错的分布式发布订阅消息系统。Kafka具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理的场景中使用。

Kafka Reader通过Kafka Java SDK读取Kafka中的数据，使用的日志服务Java SDK版本如下所示。

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.0.0</version>
</dependency>

主要涉及的Kafka SDK调用方法如下，详情请参见Kafka官方文档：

• 使用KafkaConsumer作为消息消费的客户端。

  org.apache.kafka.clients.consumer.KafkaConsumer<K,V>

• 根据unix时间戳查询Kafka点位offSet。

  Map<TopicPartition,OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition,Long> timestampsToSearch)

• 定位到开始点位offSet。

  public void seekToBeginning(Collection<TopicPartition> partitions)

• 定位到结束点位offSet。

  public void seekToEnd(Collection<TopicPartition> partitions)

• 定位到指定点位offSet。

  public void seek(TopicPartition partition,long offset)

• 客户端从服务端拉取poll数据。

  public ConsumerRecords<K,V> poll(final Duration timeout)

kafkaConfig参数说明如下。

参数	描述
fetch.min.bytes	指定消费者从broker获取消息的最小字节数，即有足够的数据时，才将其返回给消费者。
fetch.max.wait.ms	等待broker返回数据的最大时间，默认500毫秒。fetch.min.bytes和fetch.max.wait.ms先满足哪个条件，便按照该方式返回数据。
max.partition.fetch.bytes	指定broker从每个partition中返回给消费者的最大字节数，默认为1 MB。
session.timeout.ms	指定消费者不再接收服务之前，可以与服务器断开连接的时间，默认是30秒。
auto.offset.reset	消费者在读取没有偏移量或者偏移量无效的情况下（因为消费者长时间失效，包含偏移量的记录已经过时并被删除）的处理方式。默认为latest（消费者从最新的记录开始读取数据），您可以更改为earliest（消费者从起始位置读取partition的记录）。
max.poll.records	单次调用poll方法能够返回的消息数量。
key.deserializer	消息Key的反序列化方法，例如org.apache.kafka.common.serialization.StringDeserializer。
value.deserializer	数据Value的反序列化方法，例如org.apache.kafka.common.serialization.StringDeserializer。
ssl.truststore.location	SSL根证书的路径。
ssl.truststore.password	根证书Store的密码。如果是Aliyun Kafka，则配置为KafkaOnsClient。
security.protocol	接入协议，目前支持使用SASL_SSL协议接入。
sasl.mechanism	SASL鉴权方式，如果是Aliyun Kafka，使用PLAIN。

配置示例如下。

{
    "group.id": "demo_test",
    "java.security.auth.login.config": "/home/admin/kafka_client_jaas.conf",
    "ssl.truststore.location": "/home/admin/kafka.client.truststore.jks",
    "ssl.truststore.password": "KafkaOnsClient",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "ssl.endpoint.identification.algorithm": ""
}

从Kafka读取数据的JSON配置，如下所示。

{
    "type": "job",
    "steps": [
        {
            "stepType": "kafka",
            "parameter": {
                "server": "host:9093",
                "column": [
                    "__key__",
                    "__value__",
                    "__partition__",
                    "__offset__",
                    "__timestamp__",
                    "'123'",
                    "event_id",
                    "tag.desc"
                ],
                "kafkaConfig": {
                    "group.id": "demo_test"
                },
                "topic": "topicName",
                "keyType": "ByteArray",
                "valueType": "ByteArray",
                "beginDateTime": "20190416000000",
                "endDateTime": "20190416000006",
                "skipExceedRecord": "false"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {
                "print": false,
                "fieldDelimiter": ","
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": false,
            "concurrent": 1
        }
    }
}