[Improve][Sort] Import all changelog mode data ingest into Hive

[thexiay]

Description

Hive does not support CDC sink, so we could give a option to regard all changelog data as insert-only data. So we can insert into hive from canal-json，rowkind as a hive schema field.

e.g.

CREATE TABLE mysql_inlong_0505 (
	`id` BIGINT,
	`name` STRING,
        `op_type` STRING METADATA FROM 'meta.op-type',
	`log_ts` TIMESTAMP_LTZ(3) METADATA FROM 'meta.op_ts' VIRTUAL, 
	PRIMARY KEY(`id`) NOT ENFORCED
) with (
	'connector' = 'mysql-cdc-inlong',
	'hostname' = 'localhost',
	'username' = 'root',
	'password' = '123456',
	'database-name' = 'test',
	'table-name' = 'mysql_inlong_0505'
);

CREATE TABLE hive_table (
  `id` BIGINT,
  `name` STRING,
  `op_type` STRING,
) PARTITIONED BY (dt string, hr string, me string) STORED AS parquet;

INSERT INTO hive_table 
SELECT user_id, order_amount, op_type, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH'), DATE_FORMAT(log_ts, 'mm')
FROM kafka_table;

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct