flink入门

main方法

 

ConfigManager.initProp(configFilePath);

        topName = ConfigManager.initKafkaConfigV2By(args);
        // 初始化kafka信息
        // 初始化配置
        Map<String, KafkaWrite2SrConfig> configs = KafkaWrite2SrUtils.initConfigs2(topName);
        logger.info("当前环境" + ConfigManager.getProperty("hbase.keyPrefix"));

        ParameterTool parameter = ParameterTool.fromMap((Map) ConfigManager.getProp());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setGlobalJobParameters(parameter);
        // 默认4线程对齐消费者
        env.setParallelism(4);
        // 开启检查点,每10秒缓存一次
        env.enableCheckpointing(1000 * 60 * 1);// start a checkpoint every 1000 ms
        // 至少处理一次(满足条件)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000 * 60);
        env.getCheckpointConfig().setCheckpointTimeout(CheckpointParamUtil.getCheckpointTimeout(topName) * 60 * 1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 在job终止后任然保留缓存
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// 当程序关闭的时候，会触发额外的checkpoints
        String evn = parameter.get("hbase.keyPrefix");
        if (!"dev".equals(evn)) {
			env.setStateBackend(new FsStateBackend(System.getProperty("ckDir"), true));// checkpoint保存地址
        }

        String jstormEnv = ConfigManager.getProperty("jstorm.env");

        DataStreamSource<String> xxxSource = FlinkCalcUtil.getNewKafkaDataStreamSource(env, "OrderInfoDetailCkSource", new SimpleStringSchema());
        SingleOutputStreamOperator<String> ckProcess = xxxSource
        		.flatMap(new CkOrderInfoFaltMap()).setParallelism(orderInfoDetailCkSource.getParallelism())
                .keyBy(k -> k.getShipmentId()).process(new KeyedProcessFunction<String, JsonToCkOrderInfoDetailResult, String>() {
					@Override
					public void processElement(JsonToCkOrderInfoDetailResult jsonToCkOrderInfoDetailResult, KeyedProcessFunction<String, JsonToCkOrderInfoDetailResult, String>.Context ctx, Collector<String> out)
							throws Exception {
						OrderInfoDetailCkResult result = new OrderInfoDetailCkResult();
						BeanCopyUtil.copyPropertyWithSourceLongToDateAndStringToNull(jsonToCkOrderInfoDetailResult, result);
                       // 将订单状态转换为对应的各个环节字段
                        OrderStatusHandlerManager.fillOrderStatusCalc(result);
						result.setRedisTm(result.getEffectiveTime());
						out.collect(JSON.toJSONString(result));
					}
				}).name("ckProcess").setParallelism(OrderCalcUtils.loadParallelism("ckProcess", 32));
        // 算子压力比较大，资源隔离
        if (!StrUtil.equals(jstormEnv, "sit")) {
            ckProcess.slotSharingGroup("process");
        }

        ckProcess.rescale()
        .addSink(StarRocksSinkHelper.getJsonSink("ads_index_order_info_detail_new_pk", OrderInfoDetailCkResult.class,"4", false))
        .name("ckSink")
        .setParallelism(OrderCalcUtils.loadParallelism("ckSink", 8));

        DataStreamSource<String> orderInfoDetailZlSource = FlinkCalcUtil.getNewKafkaDataStreamSource(env, "OrderInfoDetailZlSource", new SimpleStringSchema());
        SingleOutputStreamOperator<String> zlProcess = orderInfoDetailZlSource
		        .flatMap(new ZlOrderInfoFaltMap()).setParallelism(orderInfoDetailZlSource.getParallelism())
		        .keyBy(k -> k.getShipmentId()).process(new KeyedProcessFunction<String, JsonToZlOrderInfoDetailResult, String>() {
					@Override
					public void processElement(JsonToZlOrderInfoDetailResult jsonToZlOrderInfoDetailResult, KeyedProcessFunction<String, JsonToZlOrderInfoDetailResult, String>.Context ctx, Collector<String> out)
							throws Exception {
						OrderInfoDetailZlResult result = new OrderInfoDetailZlResult();
				        BeanCopyUtil.copyPropertyWithSourceLongToDateAndStringToNull(jsonToZlOrderInfoDetailResult, result);
						result.setRedisTm(result.getEffectiveTime());
						out.collect(JSON.toJSONString(result));
					}
				}).name("zlProcess").setParallelism(OrderCalcUtils.loadParallelism("zlProcess", 32));
        // 算子压力比较大，资源隔离
        if (!StrUtil.equals(jstormEnv, "sit")) {
            zlProcess.slotSharingGroup("process");
        }

        zlProcess.rescale()
                .addSink(StarRocksSinkHelper.getJsonSink("xxx_pk", XXX.class,"4", false))
                .name("xxx")
                .setParallelism(OrderCalcUtils.loadParallelism("xxx", 8));;

        env.execute();

  kafka source

 

private static <T> FlinkKafkaConsumer<T> getNewConsumer(String SourceName, DeserializationSchema<T> valueDeserializer,boolean ifRerun){
        String url = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_MONITORURL));
        //<groupId:clientToken>
        String clientAndToken = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_CLIENT_ID));
        String cluster = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_CLUSTERNAME));
        //Topic token
        String tokens = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_TOKENS));
        String topic = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_TOPIC));
        String brokers = ConfigManager.getProperty(getKafkaConfigKey(SourceName, Constants.KAFKA_BROKERS));

        logger.info("kafka config: url:"+url+"---clientAndToken:"+clientAndToken+"---cluster:"+cluster+"---tokens:"+tokens+"---topic:"+topic);
        //需要topic token
        String topicToken = getTopicToken(topic, tokens);
        //获取client的groupId和token
        String[] clientSplit = clientAndToken.split(Constants.COLON);
//        String clientToken = clientSplit[1];
        String groupId = clientSplit[0];

//        String zkhost = AuthUtil.getZkServers(cluster, clientToken,groupId, url);
        // 如果配置了brokers直接使用，没有配置通过topicToken获取
        brokers = getBrokers(brokers, cluster, topicToken, url, topic);

        Properties kafkaProperties = new Properties();
        kafkaProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        kafkaProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        //自动发现新分区的间隔时间
        kafkaProperties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "12000");
        // 这里设置为 earliest，当发现新分区时，从新分区的最早位置开始消费， 建议结合业务需要合理配置，可选值有 （earliest、latest、none）
        kafkaProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        kafkaProperties.setProperty("client.dns.lookup","resolve_canonical_bootstrap_servers_only");
//        kafkaProperties.setProperty("zookeeper.connect", zkhost);
        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topic, valueDeserializer, kafkaProperties);
        /*if(ifRerun){ //重跑时从头消费
            kafkaConsumer.setStartFromEarliest();
        }else{ //根据消费者组的offset消费
            kafkaConsumer.setStartFromGroupOffsets();
        }*/
        // kafkaConsumer.setStartFromEarliest()已被禁止，现在只允许setStartFromGroupOffsets
        String starFlag = ConfigManager.getProperty("starFlag");
        if("earliest".equals(starFlag)) {// 从头消费
            kafkaConsumer.setStartFromEarliest();
        } else if("lastest".equals(starFlag)) { // 从最后消费
            kafkaConsumer.setStartFromLatest();
        } else { // 按偏移量消费
            kafkaConsumer.setStartFromGroupOffsets();
        }
        //提交偏移量开启缓存
        kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
        return kafkaConsumer;
    }

 sink sr

public static SinkFunction<String> getJsonSink(String tableName, Class clazz, String parallelism, boolean ifTranslate) {
        String jdbcUrl = ConfigManager.getProperty("starrocks.jdbc.url");
        String loadUrl = ConfigManager.getProperty("starrocks.load.url");
        String userName = ConfigManager.getProperty("starrocks.username");
        String passWord = ConfigManager.getProperty("starrocks.password");
        String databaseName = ConfigManager.getProperty("starrocks.database.name");
        String flushInterval = ConfigManager.getProperty("starrocks.flush.interval");
        String maxFilterRatio = ConfigManager.getProperty("starrocks.max.filter.ratio", "0.2");
        List<String> orderedFiledNames = getOrderedFiledNames(clazz);
        String columns = getOrderedColumns(orderedFiledNames, ifTranslate);
        String jsonPaths = getOrderedJsonPaths(orderedFiledNames);
        StarRocksSinkOptions starRocksSinkBuilder = StarRocksSinkOptions.builder().withProperty("jdbc-url", jdbcUrl).withProperty("load-url", loadUrl).withProperty("username", userName).withProperty("password", passWord).withProperty("table-name", tableName).withProperty("database-name", databaseName).withProperty("sink.properties.format", "json").withProperty("sink.properties.columns", columns).withProperty("sink.properties.jsonpaths", jsonPaths).withProperty("sink.properties.strip_outer_array", "true").withProperty("sink.properties.ignore_json_size", "true").withProperty("sink.buffer-flush.max-bytes", "94371840").withProperty("sink.properties.max_filter_ratio", maxFilterRatio).withProperty("sink.parallelism", parallelism).withProperty("sink.semantic", StarRocksSinkSemantic.AT_LEAST_ONCE.getName()).withProperty("sink.buffer-flush.interval-ms", flushInterval).withProperty("sink.properties.column_separator", "\u0001").withProperty("sink.properties.row_delimiter", "\u0002").build();
        return StarRocksSink.sink(starRocksSinkBuilder);
    }