flink多数据类型从Kafka同步到动态HDFS目录

程序

核心还是实现BucketAssigner接口，这里我们将具体逻辑抽取放到类方法中实现，目的是为了可以正常调用并访问全局配置以及对象。

.withBucketAssigner(new BucketAssigner<String, String>() {
    @Override
    public String getBucketId(String s, Context context) {
        String bucketId;  //? 定义bucketId变量
        String customBucketId = null;
        try {
        // 获取自定义的bucketId
            customBucketId = kafka2FileSys.getCustomBucketId(s);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        //? 构造bucketId并返回
        bucketId = "dtime=" + customBucketId;
        return bucketId;
    }

    @Override
    public SimpleVersionedSerializer<String> getSerializer() {
        return SimpleVersionedStringSerializer.INSTANCE;
    }
})

具体的类方法如下：

public String getCustomBucketId(String value) throws Exception {
    String dtime;
    // 从配置文件中读取数据类型以及其他配置
    String dataType = getConfigValue("datatype");  // json or text
    String dtField = getConfigValue("datetimefield");  // json datetime field
    int dtIndex = Integer.parseInt(getConfigValue("datetimefieldindex"));  // text datetime index
    String dataPatter = getConfigValue("datapatter");  // text sign
    if(dataType.equalsIgnoreCase("JSON")){
        JSONObject jsonObject = JSON.parseObject(value);
        dtime = jsonObject.getString(dtField);
    } else if (dataType.equalsIgnoreCase("TEXT")) {
        String[] valueList = value.split(dataPatter);
        dtime = valueList[dtIndex];
    }else{
        return null;
    }
    // 解析日期时间数据
    String dtFormat = getConfigValue("datetimeformat");
    DatetimeTool dtTool = new DatetimeTool(dtime, dtFormat);
    return dtTool.getDate("yyyyMMdd");
}

如果是json数据，则从一级的key中获取对应的日期时间数据；如果是csv（按照分隔符分割的文本数据）的话，从配置中加载日期时间的索引以及分隔符，然后进行解析。 然后按照对应的日期时间格式进行解析返回yyyyMMdd的日期时间字符串。 BucketAssigner接口再做前缀的拼接，例如添加dtime=的前缀，组成dtime=20230101的格式。

主要的核心代码就是上述的内容了。这个需求只是做同步，可用于同步到数仓的ODS层。具体的数据处理以及清洗的逻辑没有实现多数据类型的清洗逻辑比较麻烦，可以对各业务的数据情况以及数据类型、格式等做抽象，封装数据解析处理类，在map或者flatmap等算子中创建数据处理解析类实现解析的逻辑，也可以实现自定义的抽象接口等。 其他的工具类就不废话了。

github

GitHub地址：github.com/saberbin/Ka…

gitee地址：gitee.com/saberbin/Ka…

（因为GitHub没法正常访问，所以GitHub目前仓库是空的，gitee有完整的项目代码。）readme文档没有写，因为GitHub无法正常访问，写了也是白写。后面心情好了会写readme，push到GitHub，gitee不会再更新了[种植]。 gitee姑且是设置开源了，不知是否可以正常访问。 最后，快过年了，新春快乐。