Hive sink connector

Description

Write data to Hive.

In order to use this connector, You must ensure your spark/flink cluster already integrated hive. The tested hive version is 2.3.9.

If you use SeaTunnel Engine, You need put seatunnel-hadoop3-3.1.4-uber.jar and hive-exec-3.1.3.jar and libfb303-0.9.3.jar in $SEATUNNEL_HOME/lib/ dir.

Key features

By default, we use 2PC commit to ensure exactly-once

  • file format
    • text
    • csv
    • parquet
    • orc
    • json
  • compress codec
    • lzo

Options

name type required default value
table_name string yes -
metastore_uri string yes -
compress_codec string no none
hdfs_site_path string no -
hive_site_path string no -
hive.hadoop.conf Map no -
hive.hadoop.conf-path string no -
krb5_path string no /etc/krb5.conf
kerberos_principal string no -
kerberos_keytab_path string no -
abort_drop_partition_metadata boolean no true
common-options no -

table_name [string]

Target Hive table name eg: db1.table1, and if the source is multiple mode, you can use ${database_name}.${table_name} to generate the table name, it will replace the ${database_name} and ${table_name} with the value of the CatalogTable generate from the source.

metastore_uri [string]

Hive metastore uri

hdfs_site_path [string]

The path of hdfs-site.xml, used to load ha configuration of namenodes

hive_site_path [string]

The path of hive-site.xml

hive.hadoop.conf [map]

Properties in hadoop conf(‘core-site.xml’, ‘hdfs-site.xml’, ‘hive-site.xml’)

hive.hadoop.conf-path [string]

The specified loading path for the ‘core-site.xml’, ‘hdfs-site.xml’, ‘hive-site.xml’ files

krb5_path [string]

The path of krb5.conf, used to authentication kerberos

The path of hive-site.xml, used to authentication hive metastore

kerberos_principal [string]

The principal of kerberos

kerberos_keytab_path [string]

The keytab path of kerberos

abort_drop_partition_metadata [list]

Flag to decide whether to drop partition metadata from Hive Metastore during an abort operation. Note: this only affects the metadata in the metastore, the data in the partition will always be deleted(data generated during the synchronization process).

common options

Sink plugin common parameters, please refer to Sink Common Options for details

Example

  2.   Hive {
  3.     table_name = "default.seatunnel_orc"
  4.     metastore_uri = "thrift://namenode001:9083"
  5.   }

example 1

We have a source table like this:

  1. create table test_hive_source(
  2.      test_tinyint                          TINYINT,
  3.      test_smallint                       SMALLINT,
  4.      test_int                                INT,
  5.      test_bigint                           BIGINT,
  6.      test_boolean                       BOOLEAN,
  7.      test_float                             FLOAT,
  8.      test_double                         DOUBLE,
  9.      test_string                           STRING,
  10.      test_binary                          BINARY,
  11.      test_timestamp                  TIMESTAMP,
  12.      test_decimal                       DECIMAL(8,2),
  13.      test_char                             CHAR(64),
  14.      test_varchar                        VARCHAR(64),
  15.      test_date                             DATE,
  16.      test_array                            ARRAY<INT>,
  17.      test_map                              MAP<STRING, FLOAT>,
  18.      test_struct                           STRUCT<street:STRING, city:STRING, state:STRING, zip:INT>
  19.      )
  20. PARTITIONED BY (test_par1 STRING, test_par2 STRING);

We need read data from the source table and write to another table:

  1. create table test_hive_sink_text_simple(
  2.      test_tinyint                          TINYINT,
  3.      test_smallint                       SMALLINT,
  4.      test_int                                INT,
  5.      test_bigint                           BIGINT,
  6.      test_boolean                       BOOLEAN,
  7.      test_float                             FLOAT,
  8.      test_double                         DOUBLE,
  9.      test_string                           STRING,
  10.      test_binary                          BINARY,
  11.      test_timestamp                  TIMESTAMP,
  12.      test_decimal                       DECIMAL(8,2),
  13.      test_char                             CHAR(64),
  14.      test_varchar                        VARCHAR(64),
  15.      test_date                             DATE
  16.      )
  17. PARTITIONED BY (test_par1 STRING, test_par2 STRING);

The job config file can like this:

  1. env {
  2.   parallelism = 3
  3.   job.name="test_hive_source_to_hive"
  4. }
  6. source {
  7.   Hive {
  8.     table_name = "test_hive.test_hive_source"
  9.     metastore_uri = "thrift://ctyun7:9083"
  10.   }
  11. }
  13. sink {
  14.   # choose stdout output plugin to output data to console
  16.   Hive {
  17.     table_name = "test_hive.test_hive_sink_text_simple"
  18.     metastore_uri = "thrift://ctyun7:9083"
  19.     hive.hadoop.conf = {
  20.       bucket = "s3a://mybucket"
  21.     }
  22. }

Hive on s3

Step 1

Create the lib dir for hive of emr.

  1. mkdir -p ${SEATUNNEL_HOME}/plugins/Hive/lib

Step 2

Get the jars from maven center to the lib.

  1. cd ${SEATUNNEL_HOME}/plugins/Hive/lib
  2. wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.6.5/hadoop-aws-2.6.5.jar
  3. wget https://repo1.maven.org/maven2/org/apache/hive/hive-exec/2.3.9/hive-exec-2.3.9.jar

Step 3

Copy the jars from your environment on emr to the lib dir.

  1. cp /usr/share/aws/emr/emrfs/lib/emrfs-hadoop-assembly-2.60.0.jar ${SEATUNNEL_HOME}/plugins/Hive/lib
  2. cp /usr/share/aws/emr/hadoop-state-pusher/lib/hadoop-common-3.3.6-amzn-1.jar ${SEATUNNEL_HOME}/plugins/Hive/lib
  3. cp /usr/share/aws/emr/hadoop-state-pusher/lib/javax.inject-1.jar ${SEATUNNEL_HOME}/plugins/Hive/lib
  4. cp /usr/share/aws/emr/hadoop-state-pusher/lib/aopalliance-1.0.jar ${SEATUNNEL_HOME}/plugins/Hive/lib

Step 4

Run the case.

  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   FakeSource {
  8.     schema = {
  9.       fields {
  10.         pk_id = bigint
  11.         name = string
  12.         score = int
  13.       }
  14.       primaryKey {
  15.         name = "pk_id"
  16.         columnNames = [pk_id]
  17.       }
  18.     }
  19.     rows = [
  20.       {
  21.         kind = INSERT
  22.         fields = [1, "A", 100]
  23.       },
  24.       {
  25.         kind = INSERT
  26.         fields = [2, "B", 100]
  27.       },
  28.       {
  29.         kind = INSERT
  30.         fields = [3, "C", 100]
  31.       }
  32.     ]
  33.   }
  34. }
  36. sink {
  37.   Hive {
  38.     table_name = "test_hive.test_hive_sink_on_s3"
  39.     metastore_uri = "thrift://ip-192-168-0-202.cn-north-1.compute.internal:9083"
  40.     hive.hadoop.conf-path = "/home/ec2-user/hadoop-conf"
  41.     hive.hadoop.conf = {
  42.        bucket="s3://ws-package"
  43.     }
  44.   }
  45. }

Hive on oss

Step 1

Create the lib dir for hive of emr.

  1. mkdir -p ${SEATUNNEL_HOME}/plugins/Hive/lib

Step 2

Get the jars from maven center to the lib.

  1. cd ${SEATUNNEL_HOME}/plugins/Hive/lib
  2. wget https://repo1.maven.org/maven2/org/apache/hive/hive-exec/2.3.9/hive-exec-2.3.9.jar

Step 3

Copy the jars from your environment on emr to the lib dir and delete the conflicting jar.

  1. cp -r /opt/apps/JINDOSDK/jindosdk-current/lib/jindo-*.jar ${SEATUNNEL_HOME}/plugins/Hive/lib
  2. rm -f ${SEATUNNEL_HOME}/lib/hadoop-aliyun-*.jar

Step 4

Run the case.

  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   FakeSource {
  8.     schema = {
  9.       fields {
  10.         pk_id = bigint
  11.         name = string
  12.         score = int
  13.       }
  14.       primaryKey {
  15.         name = "pk_id"
  16.         columnNames = [pk_id]
  17.       }
  18.     }
  19.     rows = [
  20.       {
  21.         kind = INSERT
  22.         fields = [1, "A", 100]
  23.       },
  24.       {
  25.         kind = INSERT
  26.         fields = [2, "B", 100]
  27.       },
  28.       {
  29.         kind = INSERT
  30.         fields = [3, "C", 100]
  31.       }
  32.     ]
  33.   }
  34. }
  36. sink {
  37.   Hive {
  38.     table_name = "test_hive.test_hive_sink_on_oss"
  39.     metastore_uri = "thrift://master-1-1.c-1009b01725b501f2.cn-wulanchabu.emr.aliyuncs.com:9083"
  40.     hive.hadoop.conf-path = "/tmp/hadoop"
  41.     hive.hadoop.conf = {
  42.         bucket="oss://emr-osshdfs.cn-wulanchabu.oss-dls.aliyuncs.com"
  43.     }
  44.   }
  45. }

example 2

We have multiple source table like this:

  1. create table test_1(
  2. )
  3. PARTITIONED BY (xx);
  5. create table test_2(
  6. )
  7. PARTITIONED BY (xx);
  8. ...

We need read data from these source tables and write to another tables:

The job config file can like this:

  1. env {
  2.   # You can set flink configuration here
  3.   parallelism = 3
  4.   job.name="test_hive_source_to_hive"
  5. }
  7. source {
  8.   Hive {
  9.     tables_configs = [
  10.       {
  11.         table_name = "test_hive.test_1"
  12.         metastore_uri = "thrift://ctyun6:9083"
  13.       },
  14.       {
  15.         table_name = "test_hive.test_2"
  16.         metastore_uri = "thrift://ctyun7:9083"
  17.       }
  18.     ]
  19.   }
  20. }
  22. sink {
  23.   # choose stdout output plugin to output data to console
  24.   Hive {
  25.     table_name = "${database_name}.${table_name}"
  26.     metastore_uri = "thrift://ctyun7:9083"
  27.   }
  28. }

Changelog

2.2.0-beta 2022-09-26

  • Add Hive Sink Connector

2.3.0-beta 2022-10-20

  • [Improve] Hive Sink supports automatic partition repair (3133)

2.3.0 2022-12-30

  • [BugFix] Fixed the following bugs that failed to write data to files (3258)
    • When field from upstream is null it will throw NullPointerException
    • Sink columns mapping failed
    • When restore writer from states getting transaction directly failed

Next version

  • [Improve] Support kerberos authentication (3840)
  • [Improve] Added partition_dir_expression validation logic (3886)