Hive Example

Prepare to get module archive

Module archive is in the directory:inlong-sort-standalone/sort-standalone-dist/target/, the archive file is apache-inlong-sort-standalone-${project.version}-bin.tar.gz.

Prepare to modify configuration file

At first, decompress the archive file, copy three files in the directory “conf/hive/“ to the directory “conf/“.

  • conf/common.properties, common configuration of all components.
  • conf/SortClusterConfig.conf, sink configuration of all sort tasks.
  • conf/sid_hive_inlong6th_v3.conf, data source configuration example of a sort task, the file name is same with sort task name in SortClusterConfig.conf.

Example: conf/common.properties

  1. clusterId=hivev3-sz-sz1
  2. nodeId=nodeId
  3. metricDomains=Sort
  4. metricDomains.Sort.domainListeners=org.apache.inlong.sort.standalone.metrics.prometheus.PrometheusMetricListener
  5. metricDomains.Sort.snapshotInterval=60000
  6. sortChannel.type=org.apache.inlong.sort.standalone.channel.BufferQueueChannel
  7. sortSink.type=org.apache.inlong.sort.standalone.sink.hive.HiveSink
  8. sortSource.type=org.apache.inlong.sort.standalone.source.sortsdk.SortSdkSource
  10. sortClusterConfig.type=file
  11. sortClusterConfig.file=SortClusterConfig.conf
  12. sortSourceConfig.QueryConsumeConfigType=file
  13. #sortTaskId.conf
  15. #sortClusterConfig.type=manager
  16. #sortSourceConfig.QueryConsumeConfigType=manager
  17. #managerUrlLoaderType=org.apache.inlong.sort.standalone.config.loader.CommonPropertiesManagerUrlLoader
  18. #sortClusterConfig.managerUrl=http://${manager_ip:port}/api/inlong/manager/openapi/sort/getClusterConfig
  19. #sortSourceConfig.managerUrl=http://${manager_ip:port}/api/inlong/manager/openapi/sort/getSortSource

Example: conf/SortClusterConfig.conf

  1. {
  2.     "clusterName": "hivev3-sz-sz1",
  3.     "sortTasks": [{
  4.         "name": "sid_hive_inlong6th_v3",
  5.         "type": "HIVE",
  6.         "idParams": [{
  7.             "inlongGroupId": "atta",
  8.             "inlongStreamId": "0fc00000046",
  9.             "separator": "|",
  10.             "partitionIntervalMs": 3600000,
  11.             "idRootPath": "/user/hive/warehouse/t_inlong_v1_0fc00000046",
  12.             "partitionSubPath": "/{yyyyMMdd}/{yyyyMMddHH}",
  13.             "hiveTableName": "t_inlong_v1_0fc00000046",
  14.             "partitionFieldName": "dt",
  15.             "partitionFieldPattern": "yyyyMMddHH",
  16.             "msgTimeFieldPattern": "yyyy-MM-dd HH:mm:ss",
  17.             "maxPartitionOpenDelayHour": 8
  18.         }],
  19.         "sinkParams": {
  20.             "hdfsPath": "hdfs://10.160.139.123:9000",
  21.             "maxFileOpenDelayMinute": "5",
  22.             "tokenOvertimeMinute": "60",
  23.             "maxOutputFileSizeGb": "2",
  24.             "hiveJdbcUrl": "jdbc:hive2://10.160.142.179:10000",
  25.             "hiveDatabase": "default",
  26.             "hiveUsername": "hive",
  27.             "hivePassword": "hive"
  28.         }
  29.     }]
  30. }

Example: conf/sid_hive_inlong6th_v3.conf

  1. {
  2.     "sortClusterName": "hivev3-sz-sz1",
  3.     "sortTaskId": "sid_hive_inlong6th_v3",
  4.     "cacheZones": {
  5.         "pc_atta6th_sz1": {
  6.             "zoneName": "pc_atta6th_sz1",
  7.             "serviceUrl": "http://9.139.53.86:8080",
  8.             "authentication": "eyJrZXlJZCI6InB1bHNhci04MnhhN24zZWs1ZHciLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItODJ4YTduM2VrNWR3X2FkbWluIn0.D5H_j8UQk8KYWHw_mzq2HmR393SnbL5Gz7JYCANBPnI",
  9.             "topics": [{
  10.                 "topic": "pulsar-82xa7n3ek5dw/atta/atta_topic_1",
  11.                 "partitionCnt": 10,
  12.                 "topicProperties": {}
  13.             }],
  14.             "cacheZoneProperties": {},
  15.             "zoneType": "pulsar"
  16.         }
  17.     }
  18. }

Modify configuration file:conf/common.properties

ParameterRequiredDefaultValueRemark
clusterIdYNAinlong-sort-standalone cluster id
nodeIdNLocal IPCurrent node id
metricDomainsNSortdomain name of metric
metricDomains.Sort.domainListenersNorg.apache.inlong.sort.standalone.metrics.prometheus.PrometheusMetricListenerclass name list of metric listener, separated by space
metricDomains.Sort.snapshotIntervalN60000interval snapshoting metric data(millisecond)
prometheusHttpPortN8080HTTP server port of prometheus simple client
sortChannel.typeNorg.apache.inlong.sort.standalone.channel.BufferQueueChannelChannel class name
sortSink.typeYNASink class name
sortSource.typeNorg.apache.inlong.sort.standalone.source.sortsdk.SortSdkSourceSource class name
sortClusterConfig.typeNmanagerLoader source of cluster configuration data: [file,manager,UserDefinedClassName].
sortClusterConfig.fileNSortClusterConfig.confFile name in class resource when sortClusterConfig.type=file.
sortClusterConfig.managerUrlNNAThe parameter is the cluster configuration URL of InlongManager when sortClusterConfig.type=manager.
For example:http://${manager ip:port}/api/inlong/manager/openapi/sort/standalone/getClusterConfig
sortSourceConfig.QueryConsumeConfigTypeNmanagerLoader source of sort task configuration data: [file,manager,UserDefinedClassName].
Sort task configuration file is ${sortTaskId}.conf in the class resource when sortSourceConfig.QueryConsumeConfigType=file.
sortSourceConfig.managerUrlNNAThe parameter is the sort task configuration URL of InlongManager when sortClusterConfig.type=manager.
For example:http://${manager ip:port}/api/inlong/manager/openapi/sort/standalone/getSortSource

Modify configuration file: SortClusterConfig.conf

  • Get cluster configuration data from the file:SortClusterConfig.conf in classpath, it can not support online updating.
  • Get cluster configuration data from InlongManager URL, it can support online updating.
ParameterRequiredTypeDefaultValueRemark
clusterNameYStringNAinlong-sort-standalone cluster id
sortTasksYJsonArray<SortTaskConfig>NASort task list

Modify configuration: SortTaskConfig

ParameterRequiredDefaultValueRemark
nameYNAsort task name
typeYNAsort task type, for example:HIVE(“hive”), TUBE(“tube”), KAFKA(“kafka”), PULSAR(“pulsar”), ElasticSearch(“ElasticSearch”), UNKNOWN(“n”)
idParamsYNAInlong DataStream configuration
sinkParamsYNASort task parameters

Modify configuration: idParams of Hive sort task

ParameterRequiredDefaultValueRemark
inlongGroupIdYNAinlongGroupId
inlongStreamIdYNAinlongStreamId
separatorYNAseparator of Inlong datastream in data source
partitionIntervalMsN3600000partition interval(millisecond)
idRootPathYNAHDFS root path of Inlong DataStream
partitionSubPathYNApartition sub path of Inlong DataStream
hiveTableNameYNAHive table name of Inlong DataStream
partitionFieldNameNdtpartition field name of Inlong DataStream
partitionFieldPatternYNADate format of partition field value, the type have {yyyyMMdd},{yyyyMMddHH},{yyyyMMddHHmm}
msgTimeFieldPatternYNADate format of message generation time, it support Java date format
maxPartitionOpenDelayHourN8Max delay hour of partition(hour)

Modify configuration: sinkParams of Hive sort task

ParameterRequiredDefaultValueRemark
hdfsPathYNANameNode URL of HDFS
maxFileOpenDelayMinuteN5Max writing delay minutes of simple HDFS file(minute)
tokenOvertimeMinuteN60token overtime of Inlong Data Stream(minute)
maxOutputFileSizeGbN2Max file size of simple HDFS file(GB)
hiveJdbcUrlYNAJDBC URL of Hive
hiveDatabaseYNAHive database
hiveUsernameYNAHive username
hivePasswordYNAHive password

Modify configuration file: sid_hive_inlong6th_v3.conf

  • The file name include sort task name plus the postfix “.conf”.
  • Get the configuration data from the file in classpath, it can not support online updating.
  • Get the configuration data from InlongManager URL, it can support online updating.

Modify configuration: sid_hive_inlong6th_v3.conf

ParameterRequiredTypeDefaultValueRemark
sortClusterNameYStringNAinlong-sort-standalone cluster id
sortTaskIdYStringNASort task name
cacheZonesYJsonObject<String, JsonObject>NACache cluster list, Map<cacheClusterName, CacheCluster>

Modify configuration: CacheCluster

ParameterRequiredTypeDefaultValueRemark
zoneNameYStringNAcache cluster name
zoneTypeYStringNA[pulsar,tube,kafka]
serviceUrlYStringNAPulsar serviceUrl or Kafka broker list
authenticationYStringNAPulsar authentication
cacheZonePropertiesNMap<String,String>NACache consumer configuration
topicsNList<Topic>NATopic list of Cache consumer

Modify configuration: Topic

ParameterRequiredTypeDefaultValueRemark
topicYStringNAcache topic name
partitionCntYIntegerNAcache topic partition count
topicPropertiesNMap<String,String>NACache topic configuration

Start inlong-sort-standalone application

At last, execute the shell file “./bin/sort-start.sh” for starting sort-standalone, you can check the log file “sort.log”.