- flink 继承connector源码二次开发思路
- 1. idea 创建module项目
- 2. pom文件说明-以elasticsearch6为例
- 2.1 参考官方提供的连接包设置
- 2.2 maven-shade-plugin
- 2.3 elasticsearch6 二开pom配置
- 3 新增自定义连接器 my-elasticsearch6
- 3.1 创建连接器类继承Elasticsearch6DynamicSinkFactory
- 3.2 tableEnv 环境参数设置
- 3.3 获取tableEnv 环境参数
- 4 配置到org.apache.flink.table.factories.Factory
- 5. mainClass单元测试类的编写
- 6. 应用生效方式
说明：其他连接器jdbc,kafka等等二次开发思路一致
推荐：每个公司基于flink开发内部平台，一些内部的特殊场景与需求，经常需要修改源码。但是修改源码在版本更新的情况下会导致开发成本大，周期长。本方案通过继承源码的方式，通过加强，打包覆盖源码的类解决上述问题。
elasticsearch6 为案例，部分pom参考官方提供的连接包，版本号对应，
2.2 maven-shade-pluginorg.apache.flink flink-sql-connector-elasticsearch6_${scala.binary.version} ${flink.version}
- 作用：maven-shade-plugin打包
- 拷贝flink-sql-connector-elasticsearch6_${scala.binary.version}的所有配置，其余的配置根据自身情况做相应的配置
3 新增自定义连接器 my-elasticsearch6 3.1 创建连接器类继承Elasticsearch6DynamicSinkFactory${java.version} ${java.version} UTF-8 UTF-8 1.8 1.13.1 2.11 1.7.15 8.0.21 1.2.75 20.0 3.11 5.5.2 1.5.0 provided org.apache.flink flink-connector-elasticsearch6_${scala.binary.version} ${flink.version} org.apache.flink flink-connector-kafka_${scala.binary.version} ${flink.version} ${scope} slf4j-api org.slf4j org.apache.flink flink-csv ${flink.version} ${scope} org.apache.flink flink-json ${flink.version} ${scope} org.apache.flink flink-table-planner-blink_${scala.binary.version} ${flink.version} ${scope} org.apache.flink flink-table-api-java-bridge_${scala.binary.version} ${flink.version} ${scope} org.apache.flink flink-table-common ${flink.version} ${scope} org.apache.flink flink-clients_${scala.binary.version} ${flink.version} ${scope} org.apache.flink flink-table-common ${flink.version} ${scope} org.slf4j slf4j-log4j12 ${slf4j.version} ${scope} com.alibaba fastjson ${fastjson.version} cn.hutool hutool-all ${cn.hutool.all.version} org.apache.commons commons-lang3 ${apache.commons.version} com.google.guava guava ${google.guava.version} com.macasaet.fernet fernet-java8 ${macasaet.version} com.pupu.bigdata pupu-flink-connector-common 1.0.0.0 org.projectlombok lombok 1.18.10 true ${scope} org.apache.maven.plugins maven-shade-plugin 3.3.0 shade-flink package shade false *:* com.carrotsearch:hppc com.tdunning:t-digest joda-time:joda-time net.sf.jopt-simple:jopt-simple org.elasticsearch:jna org.hdrhistogram:HdrHistogram org.yaml:snakeyaml cn.hutool:hutool-all META-INF/** cn.hutool:hutool-all META-INF/** com.alibaba:fastjson META-INF/** org.elasticsearch:elasticsearch config/** modules.txt plugins.txt org/joda/** org.elasticsearch.client:elasticsearch-rest-high-level-client forbidden/** org.apache.httpcomponents:httpclient mozilla/** org.apache.lucene:lucene-analyzers-common org/tartarus/** *:* META-INF/versions/** META-INF/services/com.fasterxml.** META-INF/services/org.apache.lucene.** META-INF/services/org.elasticsearch.** META-INF/LICENSE.txt org.apache.commons org.apache.flink.elasticsearch6.shaded.org.apache.commons org.apache.http org.apache.flink.elasticsearch6.shaded.org.apache.http org.apache.lucene org.apache.flink.elasticsearch6.shaded.org.apache.lucene org.elasticsearch org.apache.flink.elasticsearch6.shaded.org.elasticsearch com.fasterxml.jackson org.apache.flink.elasticsearch6.shaded.com.fasterxml.jackson
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch6DynamicSinkFactory;
import org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchOptions;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.factories.FactoryUtil;
import pupu.flink.stream.connector.common.constants.Constant;
import pupu.flink.stream.connector.common.https.MateHttpUtils;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@Slf4j
@Internal
public class PupuElasticsearch6DynamicSinkFactory extends Elasticsearch6DynamicSinkFactory {
// 映射sql中的connector
public static final String IDENTIFIER = "my-elasticsearch-6";
@Override
public String factoryIdentifier() {
return IDENTIFIER;
}
// 新增需要校验的参数
@Override
public Set> requiredOptions() {
Set> options = super.requiredOptions();
return options;
}
// 可在该方法中做一些参数的特殊处理
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
// 环境参数
final Configuration envConfig = (Configuration) context.getConfiguration();
// tableSql 参数
Map configMap = context.getCatalogTable().getOptions();
return super.createDynamicTableSink(context);
}
}
3.2 tableEnv 环境参数设置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// tableEnv 设置换将参数
Configuration tabConfig = tableEnv.getConfig().getConfiguration();
// 设置参数
tabConfig.setString(key, value);
- 提交过程中获取环境参数
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
...
// 环境参数
final Configuration envConfig = (Configuration) context.getConfiguration();
...
return super.createDynamicTableSink(context);
}
4 配置到org.apache.flink.table.factories.Factory
配置原因：flink 集群启动会加载META-INF.service文件夹下的org.apache.flink.table.factories.Factory，动态遍历factoryIdentifier()获取到sql中指定的连接器
5. mainClass单元测试类的编写略
6. 应用生效方式- 打包，将.jar 复制到flink集群的lib目录下
- 程序包pom引用当前包使得代码生效
- 注意：检查应用报是否重复饮用flink-sql-connector-elasticsearch6_2.11 去掉；检查flink集群lib中是否存在flink-sql-connector-elasticsearch6_2.11，去掉。不去掉可能会存在改动的代码失效，运行虚拟机使用的是官网提供的类。
