sql-client.sh -i /data/flink/init.sql

-- create mongo table
CREATE TABLE flink_database.md_xxxbaseinfo (
  `_id`           String,
  `str`           String,
  `time`          TIMESTAMP_LTZ,
  `arr1`          array<ROW<`inv` String , `detail` ROW<`subSum` string , `aubSum` string , `subDetails` array<Map<String ,String >> , `aubDetails` array<Map<String ,String >> > , `invType_CN` String  >> ,
  `int`            INT
) WITH (
   'connector' = 'mongodb',
   'uri' = '<mongodb://root:mingyang100@xxx:27017>,xxx:27017,xxx:27017/admin?replicaSet=rs0&connectTimeoutMS=10000&authSource=admin',
   'database' = 'test',
   'collection' = 'xxxBaseInfo'
);

-- create hive table
SET table.sql-dialect=hive;

CREATE external TABLE ods.ods_md_xxx_xxxBaseInfo(
  `_id` string,
  `str` string, 
  `time` TIMESTAMP, 
  `arr1` array<STRUCT<`inv`:String, `detail`:STRUCT<`subSum`:String, `aubSum`:String, `subDetails`: array<Map<String,String>>, `aubDetails`: array<Map<String,String>>>, `invType_CN`: String >>,
  `int` int
  )
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.orc.OrcSerde' 
WITH SERDEPROPERTIES ( 
  'collection.delim'=',', 
  'field.delim'='\t', 
  'serialization.format'='\t') 
STORED AS ORC
LOCATION
  '<hdfs://xxx:8020/warehouse/tablespace/external/hive/ods.db/mongo/xxx/xxxBaseInfo>';

-- sync data
SET table.sql-dialect=default;
SET 'table.local-time-zone' = 'UTC';

INSERT INTO ods.ods_md_xxx_xxxBaseInfo
 SELECT *
FROM flink_database.md_xxxbaseinfo;