CREATE TABLE mongodb (
  user_id INT,
  item_id INT,
  category_id INT,
  behavior VARCHAR
) WITH (
  'connector' = 'mongodb', -- Here, it should be 'mongodb'.
  'database' = 'test', -- The name of the database.
  'collection' = 'table1',-- The data collection.
  'uri' = 'mongodb://$username:$password@$IP:$PORT,$IP:$PORT,$IP:$PORT/test?authSource=admin', -- The MongoDB connection string.
  'batchSize' = '1024' -- The number of records written per batch.
);

CREATE TABLE random_source (
  user_id INT,
  item_id INT,
  category_id INT,
  behavior VARCHAR
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '100', -- The number of records generated per second.
  'fields.user_id.kind' = 'sequence', -- Whether a bounded sequence (if yes, the output automatically stops after the sequence ends).
  'fields.user_id.start' = '1', -- The start value of the sequence.
  'fields.user_id.end' = '10000', -- The end value of the sequence.
  'fields.item_id.kind' = 'random', -- A random number without range.
  'fields.item_id.min' = '1', -- ‍The minimum random number.
  'fields.item_id.max' = '1000', -- The maximum random number.
  'fields.category_id.kind' = 'random', -- A random number without range.
  'fields.category_id.min' = '1', -- The minimum random number.
  'fields.category_id.max' = '1000', -- The maximum random number.
  'fields.behavior.length' = '5' -- The random string length.
);

CREATE TABLE mongodb (
  user_id INT,
  item_id INT,
  category_id INT,
  behavior VARCHAR
) WITH (
  'connector' = 'mongodb', -- Here, it should be 'mongodb'.
  'database' = 'test', -- The name of the database.
  'collection' = 'table1',-- The data collection.
  'uri' = 'mongodb://$username:$password@$IP:$PORT,$IP:$PORT,$IP:$PORT/test?authSource=admin', -- The MongoDB connection string.
  'batchSize' = '1024' -- The number of records written per batch.
);

insert into mongodb select * from random_source;