dws-connector-flink

简介

dws-connector-flink是在dws-client的基础上对接flink的一个工具，工具为对dwsClient的包装，整体入库能力跟dwsClient一致。目前内部只实现了DynamicTableSourceFactory、DynamicTableSinkFactory两个接口，并未实现CatalogFactory，所以不支持使用Catalog的场景。

dws-flink-connector的dws connector只支持单并发查询存量数据，暂不支持并行读取。

引入依赖

• jar包方式引入
  dws-connector-flink已发布至maven仓库，可选择最新版本使用。链接请参见：https://mvnrepository.com/artifact/com.huaweicloud.dws。

  <dependency>
      <groupId>com.huaweicloud.dws</groupId>
      <artifactId>dws-connector-flink_${scala.version}_${flink.version}</artifactId>
      <version>1.0</version>
  </dependency>

• Flink SQL方式引入

  在使用Flink SQL时需要将dws-connector-flink包及其依赖放入flink类加载目录，从1.0.3版本开始已经将带有依赖的包发布至maven仓库，使用时可直接在仓库下载使用：

  1. 选择匹配flink环境的包。

     

  2. 进入软件包详情页面。

     

  3. 选择最新版本。

     

  4. 选择查看所有文件。
     

     

     • 其中结尾为“jar-with-dependencies ”或 “all”的文件名为带依赖的包，用户均可使用。
     • 如果无对应flink版本，可用当前版本相近的版本尝试。
  5. 单击对应文件即可下载。

场景使用

• sink 模式
  为便捷使用，提供了一个DwsSink接口快速构建一个sink。示例如下：

  SinkFunction<EventLog> dwsSink = DwsSink.sink(DwsConfig.builder()
                  .withUrl("")
                  .withPassword("")
                  .withUsername("").build(), null, (DwsInvokeFunction<EventLog>) (eventLog, dwsClient, context) -> {
              dwsClient.write("test.test")
                      .setObject("id", eventLog.getGuid())
                      .setObject("name", eventLog.getEventId(), true)
                      .commit();
          });
          DataStreamSource<EventLog> source = evn.addSource(new LogSourceFunction());
          source.addSink(dwsSink);
          evn.execute();

  接口说明：

  public static <T> SinkFunction<T> sink(DwsConfig config, JSONObject context, DwsInvokeFunction<T> invoke);

  • config为dwsClient的参数，跟使用dwsClient一致。
  • context是为了便于在业务中使用缓存之类的操作提供的一个全局上下文，可在构建时指定，后续每次回调处理数据的接口便会附带此对象。
  • invoke 为一个函数接口，用于执行数据的处理：

        /**
         *  flink执行invoke时，执行数据处理回调
         * @param value 当前接收到的数据
         * @param client 根据配置构造的dwsClient
         * @param context  用户可以在构造时指定一个全局对象，每次回调时均会附带此参数，可作为全局缓存使用
         * @throws DwsClientException
         */
        void accept(T value, DwsClient client, JSONObject context) throws DwsClientException;

• SQL 模式
  1. 创建flink table：

     tableEnvironment.executeSql("create table dws_test (" +
         "   id int,\n" +
         "   name STRING,\n" +
         "   PRIMARY KEY (id) NOT ENFORCED" +
         ") WITH (\n" +
         "   'connector' = 'dws',\n" +
         "   'url' = 'jdbc:postgresql://****/gaussdb',\n" +
         "   'tablename' = 'test.test'," +
         "   'username'='dbadmin'," +
         "   'password'='***'" +
          ")");

  2. 将数据源中数据写入test表中：

     tableEnvironment.executeSql("insert into dws_test select guid as id,eventId as name from kafka_event_log")

  3. 从test表中查询数据：

     tableEnvironment.executeSql("select name from dws_test where id = 100 limit 10").print();

Flink SQL配置参数

Flink SQL中设置的PRIMARY KEY将自动映射到dws-client中的uniqueKeys。参数跟随client版本发布，参数功能与client一致，以下参数说明表示为最新参数。