MongoDB 连接器

FLINK-1.14

维表

• DDL

-- source
create table source_collection (
  id int,
  name varchar,
  proctime as proctime()
) with (
    'connector' = 'memory',
    'table.name' = 'source_collection'
  );

CREATE TABLE mongo_source
(
  _id BIGINT,
  f1 STRING,
  f2 BOOLEAN,
  f3 BINARY,
  f4 TINYINT,
  f5 SMALLINT,
  f6 INTEGER,
  f7 TIMESTAMP_LTZ(6),
  f8 TIMESTAMP_LTZ(3),
  f9 FLOAT,
  f10 DOUBLE,
  f11 DECIMAL(10, 2),
  f12 MAP<STRING, INTEGER>,
  f13 ROW<k INTEGER>,
  f14 ARRAY<STRING>,
  f15 ARRAY<ROW<k STRING>>
) WITH (
'connector' = 'mongodb',
'uri'='mongodb://user:password@localhost:53334',
'database'='test_db',
'collection'='dim_collection',
'lookup.cache.type'= 'lru',
'lookup.max-retries'='4',
'lookup.cache.max-rows'='1000',
'lookup.cache.ttl'='60000',
'lookup.cache.metric.enable'='true'
);

set 'sink_mongo.primary.keys'='_id';

-- yxx_local_mongodb 是登记好的 MongoDB 数据源
insert into yxx_local_mongodb.test_db.sink_target_collection
/*+ OPTIONS( 'sink.transaction.enable'='true'
,'sink.flush.on-checkpoint'='false',
'sink.flush.size'='1000',
'sink.bulk-flush.max-actions'='1000',
'sink.parallelism'='1') */
SELECT S.id, S.name, D._id, D.f1, D.f2 FROM source_collection
        AS S JOIN mongo_source for system_time as of S.proctime AS D ON S.id = D._id
;

• 元数据方式

需在猛犸平台登记 MongoDB 数据源

-- source
create table source_collection (
  id int,
  name varchar,
  proctime as proctime()
) with (
    'connector' = 'memory',
    'table.name' = 'source_collection'
  );

create view vi as select *, proctime() as proctime from  yxx_local_mongodb.test_db.dim_collection;

insert into yxx_local_mongodb.test_db.sink_target_collection
/*+ OPTIONS( 'sink.transaction.enable'='true'
,'sink.flush.on-checkpoint'='false',
'sink.flush.size'='1000',
'sink.bulk-flush.max-actions'='1000',
'sink.parallelism'='1') */
SELECT S.id, S.name, D._id, D.f1, D.f2 FROM source_collection
        AS S JOIN vi for system_time as of S.proctime AS D ON S.id = D._id

SINK 表

• DDL

  -- sink
  CREATE TABLE sink_target_collection (
    _id Integer,
    name String,
    primary key (_id) not enforced
  ) WITH (
    'connector' = 'mongodb',
    'uri'='mongodb://user:password@localhost:53334',
    'database'='test_db',
    'collection'='sink_target_collection',
   'sink.transaction.enable'='true',
   'sink.flush.on-checkpoint'='false',
   'sink.flush.size'='1000',
   'sink.bulk-flush.max-actions'='1000',
   'sink.parallelism'='1'
  );

• 元数据方式

set 'sink_mongo.primary.keys'='_id';

-- yxx_local_mongodb 是登记好的 MongoDB 数据源
insert into yxx_local_mongodb.test_db.sink_target_collection
/*+ OPTIONS( 'sink.transaction.enable'='true'
,'sink.flush.on-checkpoint'='false',
'sink.flush.size'='1000',
'sink.bulk-flush.max-actions'='1000',
'sink.parallelism'='1') */
SELECT id as _id, name, age FROM source_collection;

With 参数 / set 参数

• 针对某个catalog设置参数的，需要在下表的参数前添加catalog名称。如上述的: user_dim.connector.version