最近代码从11升级到了12，由于此次版本变动，废弃了很多api，导致代码不兼容。如原来的ta b leEnv中注册registerTablesink和registerTableSource方法
image.png
/**
* Registers an external {@link TableSink} with given field names and types in this
* {@link TableEnvironment}'s catalog.
* Registered sink tables can be referenced in SQL DML statements.
*
* <p>Temporary objects can shadow permanent ones. If a permanent object in a given path exists, it will
* be inaccessible in the current session. To make the permanent object available again one can drop the
* corresponding temporary object.
*
* @param name The name under which the {@link TableSink} is registered.
* @param fieldNames The field names to register with the {@link TableSink}.
* @param fieldTypes The field types to register with the {@link TableSink}.
* @param tableSink The {@link TableSink} to register.
* @deprecated Use {@link #executeSql(String) executeSql(ddl)} to register a table instead.
*/
@Deprecated
void registerTableSink(String name, String[] fieldNames, TypeInformation<?>[] fieldTypes, TableSink<?> tableSink)
说明中要求通过 executeSql(ddl) 的方法注册一个表。
既然要通过ddl创建，那就要自定义一个redis 的connector。
flink自定的connector是借助 TableFactory spi发现注册的。
所以一共需要实现两个重要的类。
image.png
image.png
1.RedisDynamicTableSourceFactory类
自定义的facory类只需要实现 DynamicTableSinkFactory 接口即可。
import com.iqiyi.talos.engine.operators.sink.dynamic.RedisDynamicTableSink;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.FactoryUtil;
import java.util.HashSet;
import java.util.Set;
/**
* @ClassName RedisDynamicTableSourceFactory
* @Description 自定义RedisSinkFactory
* @Author zwc
* @Date 2021-01-11 11:44
* @Version 1.0
**/
public class RedisDynamicTableSourceFactory implements DynamicTableSinkFactory {
// public static final ConfigOption<String> port = ConfigOptions.key("host").stringType().noDefaultValue();
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
helper.validate();
ReadableConfig options = helper.getOptions();
return new RedisDynamicTableSink(options);
}
//connector = "redis" 声明连接器名称
@Override
public String factoryIdentifier() {
return "redis";
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
Set<ConfigOption<?>> options = new HashSet();
return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
Set<ConfigOption<?>> options = new HashSet();
// options.add(port);
return options;
}
}
createDynamicTableSink: 校验ddl中 with(...) 附加的选项，并且从CatalogTable 初始化实例，将Options附加选项加载到上下文Context中。
factoryIdentifier: 连接器的名称
requiredOptions：必填参数
optionalOptions：可选参数
2.RedisDynamicTableSink类
自定义的 sink类实现 DynamicTableSink 接口
import com.iqiyi.talos.engine.job.EngineContext;
import com.iqiyi.talos.engine.operators.function.CollectionTableSinkFunction;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
/**
* @ClassName RedisDynamicTableSink
* @Description TODO
* @Author zwc
* @Date 2021-01-11 15:41
* @Version 1.0
**/
public class RedisDynamicTableSink implements DynamicTableSink {
private ReadableConfig options;
private EngineContext ctx;
public RedisDynamicTableSink(ReadableConfig options) {
this.options = options;
ctx = EngineContext.getContext();
}
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
return ChangelogMode.insertOnly();
}
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
CollectionTableSinkFunction collectionTableSinkFunction = new CollectionTableSinkFunction(ctx.getDeploySite(), ctx.getSimpifiedName());
return SinkFunctionProvider.of(collectionTableSinkFunction);
}
@Override
public DynamicTableSink copy() {
return new RedisDynamicTableSink(this.options);
}
@Override
public String asSummaryString() {
return "my_redis_sink";
}
}
getChangelogMode: 设置sink 是 insert only模式（目前只支持这种模式）
getSinkRuntimeProvider: 这里就是执行sink的具体逻辑了。这里可以直接使用flink 提供的redis-connector
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.0</version>
</dependency>
RedisMapper<RowData> stringRedisMapper = new RedisMapper<RowData>() {
//返回对应Redis命令
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.SET);
}
//从数据中获取对应Key
@Override
public String getKeyFromData(RowData rowData) {
StringData string = rowData.getString(0);
return string;
}
//从数据中获取对应Value
@Override
public String getValueFromData(RowData rowData) {
String s = rowData.toString();
return s;
由于此处我需要自定义redis数据结构，所以自己实现了RichFunction
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
CollectionTableSinkFunction collectionTableSinkFunction = new CollectionTableSinkFunction(ctx.getDeploySite(), ctx.getSimpifiedName());
return SinkFunctionProvider.of(collectionTableSinkFunction);
}
/**
* @ClassName CollectionTableSinkFunction
* @Description TODO
* @Author zwc
* @Date 2021-01-12 16:51
* @Version 1.0
**/
import com.alibaba.fastjson.JSON;
import com.iqiyi.talos.common.JedisClient;
import com.iqiyi.talos.engine.enums.DeploySite;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 这里定义了当接收到一条数据时，该如何 sink 的具体逻辑
*/
public class CollectionTableSinkFunction extends RichSinkFunction<RowData> {
private static Logger LOG = LoggerFactory.getLogger(CollectionTableSink.CollectionTableSinkFunction.class);
private static Map<String, List<String>> map = new ConcurrentHashMap<>();
private DeploySite deploySite;
private String jobName;
private static final Object lock = new Object();
public CollectionTableSinkFunction(DeploySite deploySite, String jobName) {
this.deploySite = deploySite;
this.jobName = jobName;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
LOG.info("OPEN!");
}
@Override
public void invoke(RowData row, Context context) throws Exception {
String operatorName = row.getString(0).toString();
String value = row.getString(1).toString();
//operatorName source 1
map.compute(operatorName, (k,v) -> {
if(v==null) {
v = new ArrayList<>();
v.add(value);
} else {
v.add(value);
}
return v;
});
LOG.info("put to collectionMap. [{},{}] ", operatorName, value);
LOG.info("mapSize:" + map.size());
}
public Map<String,List<String>> getContent() {
return map;
}
@Override
public void close() {
synchronized (lock) {
Map<String,List<String>> map = getContent();
String key = "TalosMockJob_" + jobName;
String value = JSON.toJSONString(map);
LOG.info("Send Mock result to redis. key:{}, value:{}", key, value);
long ttl = 24 * 3600 * 1000;
try {
JedisClient.get(deploySite.name()).setValue(key, value, ttl);
} catch (Exception e) {
}
}
}
}
3.最后一步spi
image.png
把RedisDynamicTableSourceFactory类包路径填进去即可
4.使用
ctx.getTableEnv().executeSql("create table MockJob_Data_Table (\n" +
" operatorName STRING," +
" data STRING" +
") WITH (\n" +
" 'connector' = 'redis'\n" +
")");
