命令行工具

    目前命令行工具支持 、describe 以及 create 三个命令。

    前往 inlong-manager 目录，修改 conf/application.properties 文件的以下配置即可。

    1. server.host=127.0.0.1
    2. server.port=8080
    3. default.admin.user=admin
    4. default.admin.password=inlong
    1. Usage: managerctl list [command] [command options]
    2.   Commands:
    3.     stream      Get stream main information
    4.       Usage: stream [options]
    5.         Options:
    6.         * -g, --group
    7.             inlong group id
    9.     group      Get group details
    10.       Usage: group [options]
    11.         Options:
    12.           -g, --group
    13.             inlong group id
    14.           -n, --num
    15.             the number displayed
    16.             Default: 10
    17.           -s, --status
    18.             ( CREATE | REJECTED | INITIALIZING | OPERATING | 
    19.              STARTED | FAILED | STOPPED | FINISHED | DELETED )
    21.     sink      Get sink details
    22.       Usage: sink [options]
    23.         Options:
    24.         * -g, --group
    25.             group id
    26.         * -s, --stream
    27.             stream id
    29.     source      Get source details
    30.       Usage: source [options]
    31.         Options:
    32.         * -g, --group
    33.             inlong group id
    34.         * -s, --stream
    35.             inlong stream id
    36.           -t, --type
    37.             sink type
    1. Usage: managerctl describe [command] [command options]
    2.   Commands:
    3.     stream      Get stream details
    4.       Usage: stream [options]
    5.         Options:
    6.         * -g, --group
    7.             inlong group id
    9.     group      Get group details
    10.       Usage: group [options]
    11.         Options:
    12.           -g, --group
    13.             inlong group id
    14.           -n, --num
    15.             the number displayed
    16.             Default: 10
    17.           -s, --status
    18.             Default: 0
    20.     sink      Get sink details
    21.       Usage: sink [options]
    22.         Options:
    23.         * -g, --group
    24.         * -s, --stream
    25.             inlong stream id
    27.     source      Get source details
    28.       Usage: source [options]
    29.         Options:
    30.         * -g, --group
    31.             inlong group id
    32.         * -s, --stream
    33.             inlong stream id
    34.         * -t, --type
    35.             sink type

    describe 用于展示inlong group / stream / sink / source 的详细信息，并以Json格式输出。

    create 不需要申请审核等步骤，只需将所配置信息准备在Json文件中即可。

    Json 文件主要有五个部分： groupConf 、streamConf 、streamSource 、streamSink 以及 streamFieldList

    1. {
    2.   "groupConf": {
    3.     "groupName": "test_group",
    4.     "description": "",
    5.     "proxyClusterId": "1",
    6.     "mqBaseConf": {
    7.       "type": "PULSAR",
    8.       "pulsarServiceUrl": "pulsar://127.0.0.1:6650",
    9.       "pulsarAdminUrl": "http://127.0.0.1:8080",
    10.       "tenant": "tenant",
    11.       "namespace": "namespace",
    12.       "enableCreateResource": false
    13.     },
    14.     "sortBaseConf": {
    15.       "type": "FLINK",
    16.       "serviceUrl": "127.0.0.1:8081"
    17.     },
    18.     "zookeeperEnabled": false,
    19.     "dailyRecords": 10000000,
    20.     "peakRecords": 100000,
    21.     "maxLength": 10000
    22.   },
    23.   "streamConf": {
    24.     "name": "test_stream",
    25.     "description": "",
    26.     "dataSeparator": "|",
    27.     "strictlyOrdered": true,
    28.     "topic": "topic"
    29.   },
    30.   "streamSource": {
    31.     "sourceType": "KAFKA",
    32.     "bootstrapServers": "127.0.0.1:9092",
    33.     "topic": "kafka_topic",
    34.     "sourceName": "kafka_sourceName",
    35.     "dataFormat": "json",
    36.     "autoOffsetReset": "EARLIEST"
    37.   },
    38.   "streamSink": {
    39.     "sinkType": "HIVE",
    40.     "dbName": "test_db",
    41.     "jdbcUrl": "jdbc:hive2://127.0.0.1:10000",
    42.     "authentication": {
    43.       "userName": "hive",
    44.       "password": "hive"
    45.     },
    46.     "fileFormat": "TextFile",
    47.     "dataSeparator": "|",
    48.     "dataPath": "hdfs://127.0.0.1:9000/user/hive/warehouse/test_db",
    49.     "sinkFields": [
    50.       {
    51.         "id": 0,
    52.         "fieldType": "STRING",
    53.         "fieldName": "name",
    54.         "sourceFieldType": "STRING",
    55.         "sourceFieldName": "name"
    56.       }
    58.     "tableName": "test_table",
    59.     "sinkName": "test",
    60.     "dataFormat": "json"
    61.   "streamFieldList": [
    62.     {
    63.       "id": 0,
    64.       "fieldType": "STRING",
    65.       "fieldName": "name",
    66.       "fieldComment": null,
    67.       "fieldValue": null
    68.     }
    69.   ]
    70. }

    streamSource

      1. "streamSource": {
      2.   "sourceType": "KAFKA",
      3.   "sourceName": "sourceName",
      4.   "bootstrapServers": "127.0.0.1:9092",
      5.   "topic": "kafka_topic",
      6.   "dataFormat": "json",
      7.   "autoOffsetReset": "EARLIEST"
      8.  },

    • MySQL Binlog

    • File

    streamSink

    • Hive

      1. "mqBaseConf": {
      2.   "type": "KAFKA",
      3.   "bootstrapServers": "pulsar://127.0.0.1:6650",
      4.   "topicName": "http://100.76.43.216:8080",
      5.   "dataFormat": "JSON",
      6.   "boolean": false,
      7. }

    • ClickHouse