54°

45亿数据迁移记录后续-日数据量千万级别到clickhouse

45亿数据迁移记录后续-日数据量千万级别到clickhouse

  • 相关文档地址
  • 环境 日志在一个服务器,clickhouse集群在另一个服务器。
  • 思路
    • 之前迁移数据的时候,没考虑这块,等数据整个迁移完了以后,才开始考虑,做准备的。
    • 思路1 : flume 读取 日志文件,然后使用flume 的avro模式传递日志(这里使用avro传递日志,是因为,日志服务器没有hadoop 环境,而且也是为了保持日志服务器的稳定),将读取到的日志落在hdfs,写入的规则,按照天进行分割,一天一个数据文件。然后使用waterdrop 动态传递参数日期,写一个linux 的crontab 任务,每天调用一次,每次传递昨天的日期,拉取昨天的数据,从hdfs 到 clickhouse。

      整个方案是可行的,初步测试,已经走通,但是与我们的场景不符合,因为我这边的数据,相对而言,都是实时数据,不允许出现一天的这么长时间的间隔。

    • 思路2 : 扔掉flume,使用waterdrop直接读取日志,清洗处理,然后直接到clickhous。关于waterdrop处理日志,之前博客,可翻阅进行参考。

      这个方案是错误的,因为我之前并没有正确了解waterdrop,waterdrop并不支持实时数据处理,所有针对日志文件,是不可行的, 虽然我这边已经处理,并且从日志中清洗得到了需要的数据。

    • 思路3 : flume 读取log 日志,清洗得到数据,然后写入kafka,关于kafka 与 clickhouse 的交互,clickhouse 这边已经提供了支持,相关查看,clickhouse 的外部表引擎。然后在clickhouse建议kafka 的引擎,对与kafka进行连接,然后建立物化视图,进行实时消费。

      这个方案是可行的,其中在使用clickhouse de 外部表引擎kafka 的时候,遇到了很多问题,最后查看官方的提问,以及作者的回复,得以解决。

    • 思路4 : flume 读取 数据落在kafka,然后使用waterdrop进行定时的消费kafka,offset 有kafka自动管理进行记录。

      未进行测试验证。但是waterdrop 群里有人做过。

  • 实现
    • flume 的 avro 需要注意相关。
      • flume 发送方的slink 需要填写的是接受方的ip,以及端口。我这边本地发送方是 104 ,接受方式118。 相关配置

        # 定义三大组件的名称
        agent1.sources = source1
        agent1.sinks = sink1
        agent1.channels = channel1
        

        配置 source 组件

        agent1.sources.source1.type = exec agent1.sources.source1.command = tail -F /home/logs/gps.log

        将数据流复制给多个channel

        #agent1.sources.source1.selector.type = replicating

        配置 sink 组件

        agent1.sinks.sink1.type = avro agent1.sinks.sink1.hostname = 192.168.108.118 agent1.sinks.sink1.port = 4141 agent1.sinks.sink1.request-timeout = 500000

        配置 channel 组件

        agent1.channels.channel1.type = memory agent1.channels.channel1.capacity = 100000 agent1.channels.channel1.transactionCapacity = 10000 agent1.channels.channel1.keep.alive = 80

        给 source 和 sink 绑定 channel

        agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1

        agent1.sources.source1.interceptors = filter1 search-replace1 agent1.sources.source1.interceptors.filter1.type = REGEX_FILTER agent1.sources.source1.interceptors.filter1.regex = (接收到的数据为:) agent1.sources.source1.interceptors.filter1.excludeEvents = false

        #agent1.sources.source1.interceptors.filter2.type = REGEX_FILTER #agent1.sources.source1.interceptors.filter2.regex = (接收到的数据为:) #agent1.sources.source1.interceptors.filter2.excludeEvents = false

        #agent1.sources.source1.interceptors = search-replace1 agent1.sources.source1.interceptors.search-replace1.type = search_replace agent1.sources.source1.interceptors.search-replace1.searchPattern = [^{]*(?=\{) agent1.sources.source1.interceptors.search-replace1.replaceString =

      • 这里的source 就是读取的日志文件地址

      • channel 配置的规则是 capacity > transactionCapacity > batchSize

      • 接收方的配置如下

        #定义三大组件的名称
        agent1.sources = source1
        agent1.sinks = sink1
        agent1.channels = channel1

        配置 source 组件

        agent1.sources.source1.type = avro

        要监听的 hostname 或者IP地址

        agent1.sources.source1.bind = 192.168.108.118 agent1.sources.source1.port = 4141

        配置 sink 组件

        agent1.sinks.sink1.type = hdfs agent1.sinks.sink1.hdfs.path = hdfs://cluster1/user/oracle/%Y-%m-%d #Flume在HDFS文件夹下创建新文件的固定前缀 agent1.sinks.sink1.hdfs.filePrefix = access_log #允许打开的最大文件数,如果超过这个数量,最先打开的文件会被关闭 #agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 #向 HDFS 写入内容时每次批量操作的 Event 数量 agent1.sinks.sink1.hdfs.batchSize= 5000 #文件格式,目前支持: SequenceFile 、 DataStream 、 CompressedStream 。 1. DataStream 不会压缩文件,不需要设置hdfs.codeC 2. CompressedStream 必须设置hdfs.codeC参数 agent1.sinks.sink1.hdfs.fileType = DataStream #文件写入格式 agent1.sinks.sink1.hdfs.writeFormat = Text #替换转义序列时是否使用本地时间戳 agent1.sinks.sink1.hdfs.useLocalTimeStamp = true #Flume在HDFS文件夹下创建新文件的后缀 #agent1.sinks.sink1.hdfs.fileSuffix = .txt #当前文件写入达到该大小后触发滚动创建新文件(0表示不根据文件大小来分割文件) agent1.sinks.sink1.hdfs.rollSize = 0 #当前文件写入达到该值时间后触发滚动创建新文件(0表示不按照时间来分割文件),单位:秒 agent1.sinks.sink1.hdfs.rollInterval = 0 #当前文件写入Event达到该数量后触发滚动创建新文件(0表示不根据 Event 数量来分割文件) agent1.sinks.sink1.hdfs.rollCount = 0 #允许HDFS操作的毫秒数,例如:open,write, flush, close。如果很多HFDS操作超时,这个配置应该增大。 agent1.sinks.sink1.hdfs.callTimeout = 480000 #每个HDFS sink的HDFS的IO操作线程数(例如:open,write) agent1.sinks.sink1.hdfs.threadsPoolSize = 10 #在发起一个关闭命令后,HDFS sink必须尝试重命名文件的次数 agent1.sinks.sink1.hdfs.closeTries = 0 #在几秒钟之间连续尝试关闭文件 agent1.sinks.sink1.hdfs.retryInterval = 60

        配置 channel 组件

        agent1.channels.channel1.type = memory agent1.channels.channel1.transactionCapacity = 10000 agent1.channels.channel1.keep.alive = 80 agent1.channels.channel1.capacity = 100000

        给 source 和 sink 绑定 channel

        agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1

      • 接收方的source 这里是118自己,因为它需要把自己的端口发布出去,114 才可以连接上,进行传输。这里建议填写自己的ip,不要填写localhost以及127.0.0.0

      • 发送读取日志文件的时候,需要注意,是大写的F,注意这里与 linux 的区分。

    • kafka 相关问题以及集群的搭建。
      • 这里的同事之前都搭建好了,直接使用就可以了。相关写入的flume 配置文件如下(这里只是展示 sink1 的配置)

        # 配置 sink 组件
        agent1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
        #agent1.sinks.sink1.kafka.bootstrap.servers = master:9092,slave1:9092,slave2:9092
        agent1.sinks.sink1.kafka.bootstrap.servers = 192.168.1.8.111:9092,192.168.108.112:9092,192.168.108.113:9092,192.168.108.118:9092,192.168.108.119:9092

        agent1.sinks.sink1.brokerList = master:9092,slave1:9092,slave2:9092

        agent1.sinks.sink1.topic = s agent1.sinks.sink1.kafka.flumeBatchSize = 20 agent1.sinks.sink1.kafka.producer.acks = 1 agent1.sinks.sink1.kafka.producer.linger.ms = 1 #agent1.sinks.sink1.kafka.compression.type = snappy

      • kafka 相关命令以及原理,参考上面的文档地址

    • clickhouse 相关
      • kafka 的外部表引擎

        CREATE TABLE kafka_structure_test (encrypt String,date DateTime,lon String,lat String,vec1 String,vec2 String,vec3 String,direction String,altitude String,state String,alarm String,vehicleNo String,vehicleColor String,id String,createBy String,createDt DateTime) ENGINE = Kafka SETTINGS kafka_broker_list = '192.168.108.118:9092,192.168.108.119:9092',  kafka_topic_list = 'wl_vehicle_data_clean',  kafka_group_name = 'wl_vehicle_data_up', kafka_format = 'JSONEachRow', kafka_row_delimiter = '\n', kafka_num_consumers = 1,kafka_max_block_size = 500;

      • kafka 引擎相关文档参考 文档地址

      • 物化视图建立

        CREATE MATERIALIZED VIEW consumer TO  t_plt_vehicle_location_test  AS  select id,encrypt,date as up_date,lon,createBy as create_by,createDt as create_dt,lat,vec1,vec2,vec3,direction,altitude,state,alarm, vehicleNo as vehicleno,vehicleColor as vehiclecolor from  kafka_structure_test ;
        
      • clickhouse 的测试表建立

        create table t_plt_vehicle_location_test
        (
        id String default 'MSG0',
        encrypt String default '0',
        up_date DateTime default '1970-01-01 00:00:01',
        lon String default -1,
        create_by String default 'UP_EXG_MSG_REAL_LOCATION',
        create_dt DateTime default now() ,
        lat String default -1,
        vec1 String default -1,
        vec2 String default -1,
        vec3 String default -1,
        direction String default -1,
        altitude String default -1,
        state String default -1,
        alarm String default -1,
        vehicleno String default '-1',
        vehiclecolor String default '-1',
        alarm_code String default '-1'
        )
        ENGINE = MergeTree() partition by toYYYYMM(up_date) ORDER BY
        (vehicleno,up_date) SETTINGS index_granularity = 8192
      • 我们这边在使用clickhouse 的kafka 引擎的时候,遇到了一个问题,kafka 引擎连接上kafka 以后,隔一段时间就自动掉线,连接中断无法消费,后面查看github 的相关提问,发现这是clickhouse 的相关bug, 修复好是在 ClickHouse 19.13.2.19 版本。相关链接

      • 后续其他问题,也可以在上面进行查看。报错查看clickhouse 的日志。或者在kafka 查看消费者消费的offset 即可看到 kafka 引擎是否还在连接kafka 进行消费。

  • 相关问题
    • clickhouse 的问题,clickhouse 相关对于kafka 的支持相关还不是很稳定,我们这边查看升级的版本也是19年年底才修复的问题。
    • 遇到小的问题,别人都没遇到的问题,请在三仔细查看,因为那多半是你犯了基础的低级错误,而别人没犯。所以你问别人,别人多半都不知道。
    • 细心,在细心。遇到错误,不要相信之前的逻辑判断。从新进行分析梳理。一点点的验证。

本文由【夜中孤影】发布于开源中国,原文链接:https://my.oschina.net/u/2971292/blog/3160550

全部评论: 0

    我有话说: