tencent cloud

CREATE TABLE flink_doris_source (
  name STRING,
  age INT,
  price DECIMAL(5,2),
  sale DOUBLE
  ) 
  WITH (
    'connector' = 'doris',
    'fenodes' = 'FE_IP:8030',
    'table.identifier' = 'database.table',
    'username' = 'root',
    'password' = 'password'
);

-- enable checkpoint
SET 'execution.checkpointing.interval' = '10s';
CREATE TABLE flink_doris_sink (
  name STRING,
  age INT,
  price DECIMAL(5,2),
  sale DOUBLE
  ) 
  WITH (
    'connector' = 'doris',
    'fenodes' = 'FE_IP:8030',
    'table.identifier' = 'db.table',
    'username' = 'root',
    'password' = 'password',
    'sink.label-prefix' = 'doris_label'
);

INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source

DorisOptions.Builder builder = DorisOptions.builder()
      .setFenodes("FE_IP:8030")
      .setTableIdentifier("db.table")
      .setUsername("root")
      .setPassword("password");

DorisSource<List<?>> dorisSource = DorisSourceBuilder.<List<?>>builder()
      .setDorisOptions(builder.build())
      .setDorisReadOptions(DorisReadOptions.builder().build())
      .setDeserializer(new SimpleListDeserializationSchema())
      .build();

env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "doris source").print();

// enable checkpoint
env.enableCheckpointing(10000);

DorisSink.Builder<String> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
dorisBuilder.setFenodes("FE_IP:8030")
      .setTableIdentifier("db.table")
      .setUsername("root")
      .setPassword("password");


DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-doris"); //streamload label prefix

builder.setDorisReadOptions(DorisReadOptions.builder().build())
      .setDorisExecutionOptions(executionBuilder.build())
      .setSerializer(new SimpleStringSerializer()) //serialize according to string 
      .setDorisOptions(dorisBuilder.build());


//mock string source
List<Tuple2<String, Integer>> data = new ArrayList<>();
data.add(new Tuple2<>("doris",1));
DataStreamSource<Tuple2<String, Integer>> source = env.fromCollection(data);

source.map((MapFunction<Tuple2<String, Integer>, String>) t -> t.f0 + "\\t" + t.f1)
    .sinkTo(builder.build());

// enable checkpoint
env.enableCheckpointing(10000);

//doris sink option
DorisSink.Builder<RowData> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
dorisBuilder.setFenodes("FE_IP:8030")
      .setTableIdentifier("db.table")
      .setUsername("root")
      .setPassword("password");

// json format to streamload
Properties properties = new Properties();
properties.setProperty("format", "json");
properties.setProperty("read_json_by_line", "true");
DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-doris") //streamload label prefix
              .setStreamLoadProp(properties); //streamload params

//flink rowdata‘s schema
String[] fields = {"city", "longitude", "latitude"};
DataType[] types = {DataTypes.VARCHAR(256), DataTypes.DOUBLE(), DataTypes.DOUBLE()};

builder.setDorisReadOptions(DorisReadOptions.builder().build())
      .setDorisExecutionOptions(executionBuilder.build())
      .setSerializer(RowDataSerializer.builder()    //serialize according to rowdata 
                         .setFieldNames(fields)
                         .setType("json")           //json format
                         .setFieldType(types).build())
      .setDorisOptions(dorisBuilder.build());

//mock rowdata source
DataStream<RowData> source = env.fromElements("")
  .map(new MapFunction<String, RowData>() {
      @Override
      public RowData map(String value) throws Exception {
          GenericRowData genericRowData = new GenericRowData(3);
          genericRowData.setField(0, StringData.fromString("beijing"));
          genericRowData.setField(1, 116.405419);
          genericRowData.setField(2, 39.916927);
          return genericRowData;
      }
  });

source.sinkTo(builder.build());

CREATE TABLE bitmap_sink (
dt int,
page string,
user_id int 
)
WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'test.bitmap_test',
  'username' = 'root',
  'password' = '',
  'sink.label-prefix' = 'doris_label',
  'sink.properties.columns' = 'dt,page,user_id,user_id=to_bitmap(user_id)'
)