import io.transwarp.connector.argodb.ArgoDBSinkConfig;
import io.transwarp.connector.argodb.serde.ArgoDBRecordSerializationSchema;
import io.transwarp.connector.argodb.serde.ArgoDBSimpleStringSerializer;
import io.transwarp.connector.argodb.streaming.HolodeskSinkLevelWriterFunction;
import io.transwarp.connector.argodb.table.ArgoDBSink;
import io.transwarp.connector.argodb.table.ArgodbSinkBuilder;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class ArgoDBSinkExample {

    public static void main(String[] args) {

        String metastoreUrl = System.getProperty("metastore.url");
        String tableName = System.getProperty("table.name");
        String shivaMasterGroup = System.getProperty("shiva.mastergroup");
        String kafkaTopic = System.getProperty("kafka.topic");
        String kafkaServer = System.getProperty("kafka.bootstrap.server");
        String kafkaGroupName = System.getProperty("kafka.group.name");
        String para = System.getProperty("sink.parallelism");


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(10000);

// 高级选项：

// 设置模式为精确一次 (这是默认值)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 确认 checkpoints 之间的时间会进行 500 ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);

// 允许两个连续的 checkpoint 错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

// 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// 开启实验性的 unaligned checkpoints
        env.getCheckpointConfig().enableUnalignedCheckpoints();


        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink115/checkpoints");  // 设置检查点存储路径

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(kafkaServer)
                .setTopics(kafkaTopic)
                .setGroupId(kafkaGroupName)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source").setParallelism(30);

        DataStream<String[]> dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source").setParallelism(30)
                .map(
                        new RichMapFunction<String, String[]>() {
                            @Override
                            public String[] map(String s) throws Exception {
                                return StringUtils.split(s, ",", -1);
                            }
                        }
                ).setParallelism(30);


        ArgoDBSinkConfig flinkArgoDBSinkConfig = ArgoDBSinkConfig.builder()
                .masterGroup(shivaMasterGroup)
                .tableName(tableName)
                .tmpDirectory("/tmp")
                .url(metastoreUrl)
                .enableShiva2(true)
                .useExternalAddress(false)
                .build();
        ArgoDBSimpleStringSerializer stringRecordSerializationSchema = new ArgoDBSimpleStringSerializer();

        HolodeskSinkLevelWriterFunction<String[]> argoSink = new HolodeskSinkLevelWriterFunction<>(flinkArgoDBSinkConfig, stringRecordSerializationSchema);

        dataStream.addSink(argoSink).setParallelism(Integer.parseInt(para));

        try {
            env.execute("argodb-sink");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
复制