Flink Sink应用于Clickhouse数据库

下载链接： flink-clickhouse-sink-master.zip (27.13 KB, 下载次数: 4)

2020-8-11 17:21 上传

点击文件名下载附件

Flink-Clickhouse-Sink

描述
Flink Sink用于Clickhouse数据库。 由异步Http客户端提供支持。
用于将数据加载到Clickhouse的高性能库。
它有两个加载数据的触发器：超时和缓冲区大小。



安装
Maven Central

[mw_shl_code=java,true]<dependency>
<groupId>ru.ivi.opensource</groupId>
<artifactId>flink-clickhouse-sink</artifactId>
<version>1.1.0</version>
</dependency>[/mw_shl_code]

使用Propertiesflink-clickhouse-sink使用配置属性的两个部分：common和操作员链中的每个接收器。

The common part (use like global):
clickhouse.sink.num-writers - number of writers, which build and  send requests,
clickhouse.sink.queue-max-capacity - max capacity (batches) of blank's queue,
clickhouse.sink.timeout-sec - timeout for loading data,
clickhouse.sink.retries - max number of retries,
clickhouse.sink.failed-records-path- path for failed records.
The sink part (use in chain):
clickhouse.sink.target-table - target table in Clickhouse,
clickhouse.sink.max-buffer-size- buffer size.

代码最主要的是：clickhouse-sink可处理字符串（Clickhouse插入格式，如CSV）格式的事件。 您必须将事件转换为csv格式（就像在数据库中通常插入一样）。

例如，您有event-pojo：
[mw_shl_code=java,true]class A {
   public final String str;
   public final int integer;
   
   public A(String str, int i){
       this.str = str;
       this.integer = i;
   }
}[/mw_shl_code]

然后添加记录到接收器。
您必须为Flink环境添加全局参数：

[mw_shl_code=java,true]public static String convertToCsv(A a) {
    StringBuilder builder = new StringBuilder();
    builder.append("(");
   
    // add a.str
    builder.append("'");
    builder.append(a.str);
    builder.append("', ");
   
    // add a.intger
    builder.append(String.valueOf(a.integer));
    builder.append(" )");
    return builder.toString();
}[/mw_shl_code]

并添加您的接收器，如下所示：
[mw_shl_code=java,true]// create converter
public YourEventConverter {
    String toClickHouseInsertFormat (YourEvent yourEvent){
        String chFormat = ...;
        ....
        return chFormat;
    }
}

// create props for sink
Properties props = new Properties();
props.put(ClickhouseSinkConsts.TARGET_TABLE_NAME, "your_table");
props.put(ClickhouseSinkConsts.MAX_BUFFER_SIZE, "10000");

// build chain
DataStream<YourEvent> dataStream = ...;
dataStream.map(YourEventConverter::toClickHouseInsertFormat)
          .name("convert YourEvent to Clickhouse table format")
          .addSink(new ClickhouseSink(props))
          .name("your_table clickhouse sink);[/mw_shl_code]

线路图

• reading files from "failed-records-path"
• migrate to gradle
  

作者:ivi-ru
来源:https://github.com/ivi-ru/flink-clickhouse-sink