# DGC方式如何创建Flink Hive SQL作业
#### 问题现象
在数据治理中心（DataArts Studio-DGC）创建Flink Hive SQL作业，实现从Kafka等数据源读取数据并写入Hive表。但该方式的作业创建流程与普通Flink SQL作业有所不同，需要配置Hive Catalog和特定的SQL语句。使用DGC方式如何创建Flink Hive SQL作业。
#### 解决方法
若通过DGC方式创建提交Flink Hive作业，以读Kafka写Hive作业为例，步骤如下：
1. 提前在Hive客户端中创建Hive表。例如：
   ```
   create table user_behavior_hive_tbl_no_partition(
       user_id STRING,
       item_id STRING,
       cat_id STRING,
       ts timestamp
     ) PARTITIONED BY (dy STRING, ho STRING, mi STRING) 
   stored as textfile TBLPROPERTIES (
    'partition.time-extractor.timestamp-pattern' = '$dy $ho:$mi:00',
    'sink.partition-commit.trigger' ='process-time',
    'sink.partition-commit.delay' = '0S',
    'sink.partition-commit.policy.kind' = 'metastore,success-file'
     );
   ```
   
2. 创建Flink Hive SQL作业，在DGC提交运行。SQL示例如下：
   ```
   CREATE TABLE test_kafka (
     user_id varchar,
     item_id varchar,
     cat_id varchar,
     zw_test timestamp
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'zw_test_kafka',
     'format' = 'json',
     'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
     'properties.group.id' = 'example-group1',
     'scan.startup.mode' = 'latest-offset'
   );
   CREATE CATALOG myhive WITH (
     'type' = 'hive',
     'hive-version' = '3.1.0',
     'default-database' = 'default'
   );
   use catalog myhive;
   INSERT into
     user_behavior_hive_tbl_no_partition
   SELECT
     user_id,
     item_id,
     cat_id,
     zw_test,
     DATE_FORMAT(zw_test, 'yyyy-MM-dd'),
     DATE_FORMAT(zw_test, 'HH'),
     DATE_FORMAT(zw_test, 'mm')
   FROM
     default_catalog.default_database.test_kafka;
   ```
   ![](https://support.huaweicloud.com/devg-rule-mrs/public_sys-resources/note_3.0-zh-cn.png)
   MRS 3.2.0及之前版本创建Catalog时需要在WITH参数中指定hive配置文件路径，例如：'hive-conf-dir'='obs://test-bucket/tmp/hive/config'。
   获取hive配置文件：
   1. 登录FusionInsight Manager界面，在"主页"右上方选择"下载客户端 \> 仅配置文件"，选择平台类型和下载位置后单击"确定"。
   
   2. 解压客户端文件FusionInsight_Cluster_\*_Client.tar，拷贝Hive/config下所有配置文件。