Flink二次开发

private final KeyedSerializationSchema<IN> schema;
@Override
public void invoke(KafkaTransactionState transaction, IN next, Context context) throws FlinkKafka011Exception {
    String targetTopic = schema.getTargetTopic(next);
    if (targetTopic == null) {
        targetTopic = defaultTopicId;
    }
    ...
}
（2）继承合适的父类进行二次开发


实现了KeyedSerializationSchema的类有两种：

TypeInformationKeyValueSerializationSchema<K, V>，可序列化也可反序列化，即既可用于生产者又可用于消费者
KeyedSerializationSchemaWrapper，只能序列化，只能用于生产者，传入SerializationSchema对象后可以自动调用其序列化方法进行序列化

还可以自己实现KeyedSerializationSchema接口
为了最大限度降低代码复杂度，提高执行效率，一定要选择合适的父类进行二次开发。只有序列化逻辑需要修改，因此选择实现KeyedSerializationSchema接口，重写部分方法，实现按json流内容中的ds字段值与指定的前缀组合成每条kafka信息的目标topic

JsonKeyedSerializationSchemaWrapper

public class JsonKeyedSerializationSchemaWrapper implements KeyedSerializationSchema<JSONObject> {
    private String topicPrefix;
 
    public JsonKeyedSerializationSchemaWrapper(String topicPrefix) {
        this.topicPrefix = topicPrefix;
    }
 
    @Override
    public byte[] serializeValue(JSONObject element) {
        return element.toJSONString().getBytes();
    }
 
    @Override
    public byte[] serializeKey(JSONObject element) {
        return null;
    }
 
    @Override
    public String getTargetTopic(JSONObject element) {
        return this.topicPrefix+"_"+element.getString("ds");
    }
}
（3）改造业务代码



LogParserBallEnconfEntopic

JsonKeyedSerializationSchemaWrapper serial = new JsonKeyedSerializationSchemaWrapper(
    String.format(parameterTool.get("kafka.sink.topicPrefix"),parameterTool.get("game"))
);
 
FlinkKafkaProducer011<JSONObject> kafkaResultProducer = new FlinkKafkaProducer011<>(
        parameterTool.get("kafka.sink.bootstrap.servers"),
        String.format(parameterTool.get("kafka.sink.topicPrefix"),parameterTool.get("game"))
        ,serial);
二、HdfsSink


1、按流内容分发到对应目录
flink自带的hdfs sink只支持写入到固定目录，而我们的kafka2hdfs处理逻辑要求消息要按照header_filepath字段值写入到对应的目录，类似如下形式：
/logs/ball/json/Chat/ds=2018-11-12//logs/ball/json/Chat/status-1/ds=2018-11-12//logs/ball/json/status-2-CheckWg/ds=2018-10-31/
具体实现思路如下：
（1）由源码可知BucketingSink类的setBucketer(Bucketer<T> bucketer)方法确定要写入的文件目录
（2）继承合适的父类进行二次开发
实现了Bucketer接口的类有以下两种：


DateTimeBucketer 写入到固定目录的桶内，桶是按给定日期格式生成的





BasePathBucketer 写入到固定目录








我们所要写入的目录不需要分桶，因此继承BasePathBucketer类，重写部分方法

HdfsBasePathBucketer

public class HdfsBasePathBucketer extends BasePathBucketer<JSONObject> {
    private static final long serialVersionUID = 1L;
 
    @Override
    public Path getBucketPath(Clock clock, Path basePath, JSONObject element) {
        String header_filepath = element.getString("header_filepath");
        return super.getBucketPath(clock, new Path(basePath+"/"+header_filepath), element);
    }
}
 （3）改造业务代码
Bucketer bucketer = new HdfsBasePathBucketer();
hdfsSink.setBucketer(bucketer);