前言

设计一种写入框架实现类似于logstash般可配置化消费kfk数据到异构数据源（clickhouse），同时能实现数据的去重。相当于是做了一个java去重版本的cksinker。其他整体流程不变

架构

系统流程设计

注意：

1，只做消费kafka写clickhouse功能。

2，process只做去重功能。

3，数据一致性只能保证，至少一次。

4，当前版本只实现了数据快照到本地文件的功能，当前服务只能支持单进程去重一致性，因此可能存在单节点问题，如果需要多节点多进程只能将Bloomfilter放到远程（redis:rebloom，其他的第三方存储考察来看暂时都不行），且要免快照化才能实现去重bloomfilter数据的一致性。

5，由于bloomfilter存在误判，在判断存在的情况下可能是错误的，但是判断不存在的情况下，肯定是不存在的，那这样可能会导致数据丢失，所以一定要控制好错误比例，如果后期出现确实大量的数据丢失，那只能改下bitmap位图的形式了，或者（Roaringbitmaps更好的压缩性能），不过要花费大量的内存成本，以及极大降低系统的稳定性。不过也可以考虑类似于logstash那样的死队列的思路。

配置参数说明

{
    "job": {
        "setting": {
            "channel": 10,
            "data.dir":"/ssd1/services/jchsinker",
            "keep.alive.timeout.mspoll.duration.ms":1000,
            "name":"jchsinker_demo1"
        },
        "content": {
            "inputs": [{
                "name": "kafkareader",
                "parameter": {
                    "bootstrap.servers": "host:6667",
                    "topic": "test-topic-in",
                    "codec": "json",
                    "auto.offset.reset": "earliest",
                    "value.deserializer": "org.apache.kafka.common.serialization.StringDeserializer",
                    "key.deserializer": "org.apache.kafka.common.serialization.StringDeserializer",
                    "poll.interval.ms": "5000",
                    "max.poll.records":1000,
                    "group.id": "chpipeline_groupid_001",
                    "security.protocol": "SASL_PLAINTEXT",
                    "sasl.mechanism": "PLAIN",
                    "sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xxxx\" password=\"xxxx\";",
                    "table": {
                        "name": "kfk_demo_view",
                        "column": [{
                                "alias": "beginTime",
                                "name": "BeginTime"
                            },
                            {
                                "alias": "endTime",
                                "name": "EndTime"
                            },
                            {
                                "alias": "chargingItem",
                                "name": "ChargingItem"
                            },
                            {
                                "alias": "chargingUnit",
                                "name": "ChargingUnit"
                            },
                            {
                                "alias": "chargingValue",
                                "name": "ChargingValue"
                            },
                            {
                                "alias": "inValue",
                                "name": "InValue"
                            },
                            {
                                "alias": "outValue",
                                "name": "OutValue"
                            },
                            {
                                "alias": "model",
                                "name": "Model"
                            },
                            {
                                "alias": "nodeCode",
                                "name": "NodeCode"
                            },
                            {
                                "alias": "recordType",
                                "name": "RecordType"
                            },
                            {
                                "alias": "resourceId",
                                "name": "ResourceId"
                            },
                            {
                                "alias": "resourceType",
                                "name": "ResourceType"
                            },
                            {
                                "alias": "userId",
                                "name": "UserId"
                            }
                        ]
                    }
                }
            }],
            "processors": [{
                "name": "memduplicate",
                "enable": true,
                "parameter": {
                    "duplicate.key": [
                        "resourceId",
                        "userId",
                        "chargingItem",
                        "beginTime"
                    ],
                    "engine": {
                        "name": "bloomfilter",
                        "time.column": "beginTime",
                        "time.data.type": "string_t",
                        "time.format":"yyyy-MM-dd HH:mm:ss",
                        "cardinality": 1000000,
                        "fpp": 0.01,
                        "instances": 200,
                        "cache.day": 1,
                        "snapshot.interval.ms":60
                    },
                    "data.source": {
                        "type": "file"
                    }
                }
            }],
            "outputs": [{
                "name": "clickhousewriter",
                "parameter": {
                    "batch.size": 10000,
                    "flush.interval": 3000,
                    "connection": {
                        "jdbc.url": "jdbc:clickhouse://ip:port/test_db",
                        "table": "test_table_all"
                    },
                    "retry.times": 3,
                    "username": "xxxx",
                    "password": "xxxx"
                }
            }],
            "metrics": [{
                "Prometheus": {
                    "host": "127.0.0.1",
                    "interval": "5 SECONDS",
                    "class": "cn.chinatelecom.jchsinker.metrics.promethues.PrometheusReporter",
                    "port": "29091"
                }
            }]
        }
    }
}

参数详解

参数

类型

必须?

描述

1000

1000

9249

数据完整性保证

数据的完整性主要分量种exactly-once（精确一次语义）和at-least-once（至少一次语义）。

完整性消费架构

去重逻辑处理细节

消费快照文件

register.json

{ 
    ##记录成功写入到clickhouse数据库的offset
    "consumedOffset": {
        "demo1-in#0": 546983,
        "demo1-in#9": 543723,
        "demo1-in#6": 544547,
        "demo1-in#5": 544166,
        "demo1-in#8": 541308,
        "demo1-in#7": 545393,
        "demo1-in#2": 541287,
        "demo1-in#1": 545401,
        "demo1-in#4": 544586,
        "demo1-in#3": 543736
    },
   ##更新时间
    "updateTime": 1636106394830,
   ##队列中余留的数据长度
   "outputBatchQueueRemain": {
        "3#": 60,
        "2#": 69,
        "1#": 29,
        "0#": 32,
        "9#": 62,
        "8#": 22,
        "7#": 22,
        "6#": 64,
        "5#": 41,
        "4#": 29
    },
    ##去重任务去重后的最大offset
    "dumpHighWatermarkOffset": {
        "0#": 547014,
        "9#": 543784,
        "5#": 544206,
        "6#": 544610,
        "7#": 545414,
        "8#": 541329,
        "1#": 545429,
        "2#": 541355,
        "3#": 543795,
        "4#": 544614
    }
}

数据完整性监控

curl ip:port/metrics

输入数据量：TASK_pipeline_numRecordsIn

输出数据量：TASK_pipeline_numRecordsOut 

判断两个数据是否相等。

完整性保障流程

为保证整个流程数据的完整性

0，数据消费一个线程处理一个kafka 分区，kafka offset消费单调递增。

1，定时记录快照文件，包括bloomfilter文件信息，不同任务的状态信息register.json。

2，在进程奔溃的时候同样需要记录状态信息到本地。

3，在进程重启或者奔溃的情况下，需要按记录的状态文件（register.json）做replay。