BROKER LOAD

description

  1. Broker load will load data into Doris via Broker.
  2. Use `show broker;` to see the Broker deployed in cluster.
  4. Support following data sources:
  6. 1. Baidu HDFS: hdfs for Baidu. Only be used inside Baidu.
  7. 2. Baidu AFS: afs for Baidu. Only be used inside Baidu.
  8. 3. Baidu Object Storage(BOS): BOS on Baidu Cloud.
  9. 4. Apache HDFS.
  10. 5. Amazon S3：Amazon S3。

Syntax:

  1. LOAD LABEL load_label
  2. (
  3. data_desc1[, data_desc2, ...]
  4. )
  5. WITH [BROKER broker_name | S3]
  6. [load_properties]
  7. [opt_properties];
  9. 1. load_label
  11.     Unique load label within a database.
  12.     syntax:
  13.     [database_name.]your_label
  15. 2. data_desc
  17.     To describe the data source. 
  18.     syntax:
  19.         [MERGE|APPEND|DELETE]
  20.         DATA INFILE
  21.         (
  22.         "file_path1"[, file_path2, ...]
  23.         )
  24.         [NEGATIVE]
  25.         INTO TABLE `table_name`
  26.         [PARTITION (p1, p2)]
  27.         [COLUMNS TERMINATED BY "column_separator"]
  28.         [FORMAT AS "file_type"]
  29.         [(column_list)]
  30.         [PRECEDING FILTER predicate]
  31.         [SET (k1 = func(k2))]
  32.         [WHERE predicate] 
  33.         [DELETE ON label=true]
  34.         [read_properties]
  36.     Explain:
  37.         file_path: 
  39.         File path. Support wildcard. Must match to file, not directory. 
  41.         PARTITION:
  43.         Data will only be loaded to specified partitions. Data out of partition's range will be filtered. If not specifed, all partitions will be loaded.
  45.         NEGATIVE:
  47.         If this parameter is specified, it is equivalent to importing a batch of "negative" data to offset the same batch of data loaded before.
  49.         This parameter applies only to the case where there are value columns and the aggregation type of value columns is only SUM.
  51.         column_separator: 
  53.         Used to specify the column separator in the import file. Default is `\t`.
  54.         If the character is invisible, it needs to be prefixed with `\\x`, using hexadecimal to represent the separator.
  56.         For example, the separator `\x01` of the hive file is specified as `\\ x01`
  58.         file_type: 
  60.         Used to specify the type of imported file, such as parquet, orc, csv. Default values are determined by the file suffix name. 
  62.         column_list:
  64.         Used to specify the correspondence between columns in the import file and columns in the table.
  66.         When you need to skip a column in the import file, specify it as a column name that does not exist in the table.
  68.         syntax:
  69.         (col_name1, col_name2, ...)
  71.         PRECEDING FILTER predicate:
  73.         Used to filter original data. The original data is the data without column mapping and transformation. The user can filter the data before conversion, select the desired data, and then perform the conversion.
  75.         SET:
  77.         If this parameter is specified, a column of the source file can be transformed according to a function, and then the transformed result can be loaded into the table. The grammar is `column_name = expression`. Some examples are given to help understand.
  79.         Example 1: There are three columns "c1, c2, c3" in the table. The first two columns in the source file correspond in turn (c1, c2), and the last two columns correspond to c3. Then, column (c1, c2, tmp_c3, tmp_c4) SET (c3 = tmp_c3 + tmp_c4) should be specified.
  81.         Example 2: There are three columns "year, month, day" in the table. There is only one time column in the source file, in the format of "2018-06-01:02:03". Then you can specify columns (tmp_time) set (year = year (tmp_time), month = month (tmp_time), day = day (tmp_time)) to complete the import.
  83.         WHERE:
  85.         After filtering the transformed data, data that meets where predicates can be loaded. Only column names in tables can be referenced in WHERE statements.
  87.         merge_type:
  89.         The type of data merging supports three types: APPEND, DELETE, and MERGE. APPEND is the default value, which means that all this batch of data needs to be appended to the existing data. DELETE means to delete all rows with the same key as this batch of data. MERGE semantics Need to be used in conjunction with the delete condition, which means that the data that meets the delete on condition is processed according to DELETE semantics and the rest is processed according to APPEND semantics
  91.         delete_on_predicates:
  93.         Only used when merge type is MERGE
  95.         read_properties:
  97.         Used to specify some special parameters.
  98.         Syntax：
  99.         [PROPERTIES ("key"="value", ...)]
  101.         You can specify the following parameters:
  103.           line_delimiter： Used to specify the line delimiter in the load file. The default is `\n`. You can use a combination of multiple characters as the column separator.
  105.           fuzzy_parse： Boolean type, true to indicate that parse json schema as the first line, this can make import more faster,but need all key keep the order of first line, default value is false. Only use for json format.
  107.           jsonpaths: There are two ways to import json: simple mode and matched mode.
  108.             simple mode: it is simple mode without setting the jsonpaths parameter. In this mode, the json data is required to be the object type. For example:
  109.             {"k1": 1, "k2": 2, "k3": "hello"}, where k1, k2, k3 are column names.
  111.             matched mode: the json data is relatively complex, and the corresponding value needs to be matched through the jsonpaths parameter.
  113.           strip_outer_array: Boolean type, true to indicate that json data starts with an array object and flattens objects in the array object, default value is false. For example：
  114.             [
  115.               {"k1" : 1, "v1" : 2},
  116.               {"k1" : 3, "v1" : 4}
  117.             ]
  118.           if strip_outer_array is true, and two rows of data are generated when imported into Doris.
  120.           json_root: json_root is a valid JSONPATH string that specifies the root node of the JSON Document. The default value is "".
  122.           num_as_string: Boolean type, true means that when parsing the json data, it will be converted into a number type and converted into a string, and then it will be imported without loss of precision.
  124. 3. broker_name
  126.     The name of the Broker used can be viewed through the `show broker` command.
  128. 4. load_properties
  130.     Used to provide Broker access to data sources. Different brokers, and different access methods, need to provide different information.
  132.     1. Baidu HDFS/AFS
  134.         Access to Baidu's internal hdfs/afs currently only supports simple authentication, which needs to be provided:
  136.         username: hdfs username
  137.         password: hdfs password
  139.     2. BOS
  141.         bos_endpoint.
  142.         bos_accesskey: cloud user's accesskey
  143.         bos_secret_accesskey: cloud user's secret_accesskey
  145.     3. Apache HDFS
  147.         Community version of HDFS supports simple authentication, Kerberos authentication, and HA configuration.
  149.         Simple authentication:
  150.         hadoop.security.authentication = simple (default)
  151.         username: hdfs username
  152.         password: hdfs password
  154.         kerberos authentication: 
  155.         hadoop.security.authentication = kerberos
  156.         kerberos_principal: kerberos's principal
  157.         kerberos_keytab: path of kerberos's keytab file. This file should be able to access by Broker
  158.         kerberos_keytab_content: Specify the contents of the KeyTab file in Kerberos after base64 encoding. This option is optional from the kerberos_keytab configuration. 
  160.         namenode HA: 
  161.         By configuring namenode HA, new namenode can be automatically identified when the namenode is switched
  162.         dfs.nameservices: hdfs service name, customize, eg: "dfs.nameservices" = "my_ha"
  163.         dfs.ha.namenodes.xxx: Customize the name of a namenode, separated by commas. XXX is a custom name in dfs. name services, such as "dfs. ha. namenodes. my_ha" = "my_nn"
  164.         dfs.namenode.rpc-address.xxx.nn: Specify RPC address information for namenode, where NN denotes the name of the namenode configured in dfs.ha.namenodes.xxxx, such as: "dfs.namenode.rpc-address.my_ha.my_nn"= "host:port"
  165.         dfs.client.failover.proxy.provider: Specify the provider that client connects to namenode by default: org. apache. hadoop. hdfs. server. namenode. ha. Configured Failover ProxyProvider.
  166.     4. Amazon S3
  168.         fs.s3a.access.key：AmazonS3的access key
  169.         fs.s3a.secret.key：AmazonS3的secret key
  170.         fs.s3a.endpoint：AmazonS3的endpoint 
  171.     5. If using the S3 protocol to directly connect to the remote storage, you need to specify the following attributes 
  173.         (
  174.             "AWS_ENDPOINT" = "",
  175.             "AWS_ACCESS_KEY" = "",
  176.             "AWS_SECRET_KEY"="",
  177.             "AWS_REGION" = ""
  178.         )
  179.     6. if using load with hdfs, you need to specify the following attributes 
  180.         (
  181.             "fs.defaultFS" = "",
  182.             "hdfs_user"="",
  183.             "dfs.nameservices"="my_ha",
  184.             "dfs.ha.namenodes.xxx"="my_nn1,my_nn2",
  185.             "dfs.namenode.rpc-address.xxx.my_nn1"="host1:port",
  186.             "dfs.namenode.rpc-address.xxx.my_nn2"="host2:port",
  187.             "dfs.client.failover.proxy.provider.xxx"="org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  188.         )
  189.         fs.defaultFS: defaultFS
  190.         hdfs_user: hdfs user
  191.         namenode HA：
  192.         By configuring namenode HA, new namenode can be automatically identified when the namenode is switched
  193.         dfs.nameservices: hdfs service name, customize, eg: "dfs.nameservices" = "my_ha"
  194.         dfs.ha.namenodes.xxx: Customize the name of a namenode, separated by commas. XXX is a custom name in dfs. name services, such as "dfs. ha. namenodes. my_ha" = "my_nn"
  195.         dfs.namenode.rpc-address.xxx.nn: Specify RPC address information for namenode, where NN denotes the name of the namenode configured in dfs.ha.namenodes.xxxx, such as: "dfs.namenode.rpc-address.my_ha.my_nn"= "host:port"
  196.         dfs.client.failover.proxy.provider: Specify the provider that client connects to namenode by default: org. apache. hadoop. hdfs. server. namenode. ha. Configured Failover ProxyProvider.
  198. 4. opt_properties
  200.     Used to specify some special parameters. 
  201.     Syntax:
  202.     [PROPERTIES ("key"="value", ...)]
  204.     You can specify the following parameters:
  206.     timout: Specifies the timeout time for the import operation. The default timeout is 4 hours per second.
  208.     max_filter_ratio: Data ratio of maximum tolerance filterable (data irregularity, etc.). Default zero tolerance.
  210.     exc_mem_limit: Memory limit. Default is 2GB. Unit is Bytes.
  212.     strict_mode: Whether the data is strictly restricted. The default is false.
  214.     timezone: Specify time zones for functions affected by time zones, such as strftime/alignment_timestamp/from_unixtime, etc. See the documentation for details. If not specified, use the "Asia/Shanghai" time zone.
  216.     send_batch_parallelism: Used to set the default parallelism for sending batch, if the value for parallelism exceed `max_send_batch_parallelism_per_job` in BE config, then the coordinator BE will use the value of `max_send_batch_parallelism_per_job`.
  218. 5. Load data format sample
  220.     Integer（TINYINT/SMALLINT/INT/BIGINT/LARGEINT）: 1, 1000, 1234
  221.     Float（FLOAT/DOUBLE/DECIMAL）: 1.1, 0.23, .356
  222.     Date（DATE/DATETIME）: 2017-10-03, 2017-06-13 12:34:03. 
  223.     (Note: If it's in other date formats, you can use strftime or time_format functions to convert in the import command)
  225.     String（CHAR/VARCHAR）: "I am a student", "a"
  226.     NULL: \N

example

  1. 1. Load a batch of data from HDFS, specify timeout and filtering ratio. Use the broker with the plaintext ugi my_hdfs_broker. Simple authentication.
  3.     LOAD LABEL example_db.label1
  4.     (
  5.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  6.     INTO TABLE `my_table`
  7.     )
  8.     WITH BROKER my_hdfs_broker
  9.     (
  10.     "username" = "hdfs_user",
  11.     "password" = "hdfs_passwd"
  12.     )
  13.     PROPERTIES
  14.     (
  15.     "timeout" = "3600",
  16.     "max_filter_ratio" = "0.1"
  17.     );
  19.     Where hdfs_host is the host of the namenode and hdfs_port is the fs.defaultFS port (default 9000)
  21. 2. Load a batch of data from AFS contains multiple files. Import different tables, specify separators, and specify column correspondences.
  23.     LOAD LABEL example_db.label2
  24.     (
  25.     DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file1")
  26.     INTO TABLE `my_table_1`
  27.     COLUMNS TERMINATED BY ","
  28.     (k1, k3, k2, v1, v2),
  29.     DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file2")
  30.     INTO TABLE `my_table_2`
  31.     COLUMNS TERMINATED BY "\t"
  32.     (k1, k2, k3, v2, v1)
  33.     )
  34.     WITH BROKER my_afs_broker
  35.     (
  36.     "username" = "afs_user",
  37.     "password" = "afs_passwd"
  38.     )
  39.     PROPERTIES
  40.     (
  41.     "timeout" = "3600",
  42.     "max_filter_ratio" = "0.1"
  43.     );
  46. 3. Load a batch of data from HDFS, specify hive's default delimiter \\x01, and use wildcard * to specify all files in the directory. Use simple authentication and configure namenode HA at the same time
  48.     LOAD LABEL example_db.label3
  49.     (
  50.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/*")
  51.     INTO TABLE `my_table`
  52.     COLUMNS TERMINATED BY "\\x01"
  53.     )
  54.     WITH BROKER my_hdfs_broker
  55.     (
  56.     "username" = "hdfs_user",
  57.     "password" = "hdfs_passwd",
  58.     "dfs.nameservices" = "my_ha",
  59.     "dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
  60.     "dfs.namenode.rpc-address.my_ha.my_namenode1" = "nn1_host:rpc_port",
  61.     "dfs.namenode.rpc-address.my_ha.my_namenode2" = "nn2_host:rpc_port",
  62.     "dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  63.     )
  65. 4. Load a batch of "negative" data from HDFS. Use Kerberos authentication to provide KeyTab file path.
  67.     LOAD LABEL example_db.label4
  68.     (
  69.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/old_file)
  70.     NEGATIVE
  71.     INTO TABLE `my_table`
  72.     COLUMNS TERMINATED BY "\t"
  73.     )
  74.     WITH BROKER my_hdfs_broker
  75.     (
  76.     "hadoop.security.authentication" = "kerberos",
  77.     "kerberos_principal"="doris@YOUR.COM",
  78.     "kerberos_keytab"="/home/palo/palo.keytab"
  79.     )
  81. 5. Load a batch of data from HDFS, specify partition. At the same time, use Kerberos authentication mode. Provide the KeyTab file content encoded by base64.
  83.     LOAD LABEL example_db.label5
  84.     (
  85.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  86.     INTO TABLE `my_table`
  87.     PARTITION (p1, p2)
  88.     COLUMNS TERMINATED BY ","
  89.     (k1, k3, k2, v1, v2)
  90.     )
  91.     WITH BROKER my_hdfs_broker
  92.     (
  93.     "hadoop.security.authentication"="kerberos",
  94.     "kerberos_principal"="doris@YOUR.COM",
  95.     "kerberos_keytab_content"="BQIAAABEAAEACUJBSURVLkNPTQAEcGFsbw"
  96.     )
  98. 6. Load a batch of data from BOS, specify partitions, and make some transformations to the columns of the imported files, as follows:
  100.    Table schema: 
  101.     k1 varchar(20)
  102.     k2 int
  104.     Assuming that the data file has only one row of data:
  106.     Adele,1,1
  108.     The columns in the data file correspond to the columns specified in the load statement:
  110.     k1,tmp_k2,tmp_k3
  112.     transform as: 
  114.     1) k1: unchanged
  115.     2) k2: sum of tmp_k2 and tmp_k3
  117.     LOAD LABEL example_db.label6
  118.     (
  119.     DATA INFILE("bos://my_bucket/input/file")
  120.     INTO TABLE `my_table`
  121.     PARTITION (p1, p2)
  122.     COLUMNS TERMINATED BY ","
  123.     (k1, tmp_k2, tmp_k3)
  124.     SET (
  125.       k2 = tmp_k2 + tmp_k3
  126.     )
  127.     )
  128.     WITH BROKER my_bos_broker
  129.     (
  130.     "bos_endpoint" = "http://bj.bcebos.com",
  131.     "bos_accesskey" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
  132.     "bos_secret_accesskey"="yyyyyyyyyyyyyyyyyyyy"
  133.     )
  135. 7. Load data into tables containing HLL columns, which can be columns in tables or columns in data
  137.     If there are 4 columns in the table are (id, v1, v2, v3). The v1 and v2 columns are hll columns. The imported source file has 3 columns, where the first column in the table = the first column in the source file, and the second and third columns in the table are the second and third columns in the source file, and the third column in the table is transformed. The four columns do not exist in the source file.
  138.     Then (column_list) declares that the first column is id, and the second and third columns are temporarily named k1, k2.
  140.     In SET, the HLL column in the table must be specifically declared hll_hash. The V1 column in the table is equal to the hll_hash (k1) column in the original data.The v3 column in the table does not have a corresponding value in the original data, and empty_hll is used to supplement the default value.
  142.     LOAD LABEL example_db.label7
  143.     (
  144.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  145.     INTO TABLE `my_table`
  146.     PARTITION (p1, p2)
  147.     COLUMNS TERMINATED BY ","
  148.     (id, k1, k2)
  149.     SET (
  150.       v1 = hll_hash(k1),
  151.       v2 = hll_hash(k2),
  152.       v3 = empty_hll()
  153.     )
  154.     )
  155.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  157.     LOAD LABEL example_db.label8
  158.     (
  159.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  160.     INTO TABLE `my_table`
  161.     PARTITION (p1, p2)
  162.     COLUMNS TERMINATED BY ","
  163.     (k1, k2, tmp_k3, tmp_k4, v1, v2)
  164.     SET (
  165.       v1 = hll_hash(tmp_k3),
  166.       v2 = hll_hash(tmp_k4)
  167.     )
  168.     )
  169.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  171. 8. Data in load Parquet file specifies FORMAT as parquet. By default, it is judged by file suffix.
  173.     LOAD LABEL example_db.label9
  174.     (
  175.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  176.     INTO TABLE `my_table`
  177.     FORMAT AS "parquet"
  178.     (k1, k2, k3)
  179.     )
  180.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  182. 9. Extract partition fields in file paths
  184.     If necessary, partitioned fields in the file path are resolved based on the field type defined in the table, similar to the Partition Discovery function in Spark.
  186.     LOAD LABEL example_db.label10
  187.     (
  188.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/*/*")
  189.     INTO TABLE `my_table`
  190.     FORMAT AS "csv"
  191.     (k1, k2, k3)
  192.     COLUMNS FROM PATH AS (city, utc_date)
  193.     SET (uniq_id = md5sum(k1, city))
  194.     )
  195.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  197.     Directory `hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing` contains following files:
  199.     [hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/utc_date=2019-06-26/0000.csv, hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/utc_date=2019-06-26/0001.csv, ...]
  201.     Extract city and utc_date fields in the file path
  203. 10. To filter the load data, columns whose K1 value is greater than K2 value can be imported.
  205.     LOAD LABEL example_db.label10
  206.     (
  207.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  208.     INTO TABLE `my_table`
  209.     where k1 > k2
  210.     );
  212. 11. Extract date partition fields in file paths, and date time include %3A (in hdfs path, all ':' will be replaced by '%3A')
  214.     Assume we have files:
  216.     /user/data/data_time=2020-02-17 00%3A00%3A00/test.txt
  217.     /user/data/data_time=2020-02-18 00%3A00%3A00/test.txt
  219.     Table schema is:
  220.     data_time DATETIME,
  221.     k2        INT,
  222.     k3        INT
  224.     LOAD LABEL example_db.label12
  225.     (
  226.      DATA INFILE("hdfs://host:port/user/data/*/test.txt") 
  227.      INTO TABLE `tbl12`
  228.      COLUMNS TERMINATED BY ","
  229.      (k2,k3)
  230.      COLUMNS FROM PATH AS (data_time)
  231.      SET (data_time=str_to_date(data_time, '%Y-%m-%d %H%%3A%i%%3A%s'))
  232.     ) 
  233.     WITH BROKER "hdfs" ("username"="user", "password"="pass");
  235. 12. Load a batch of data from HDFS, specify timeout and filtering ratio. Use the broker with the plaintext ugi my_hdfs_broker. Simple authentication. delete the data when v2 >100, other append
  237.     LOAD LABEL example_db.label1
  238.     (
  239.     MERGE DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  240.     INTO TABLE `my_table`
  241.     COLUMNS TERMINATED BY "\t"
  242.     (k1, k2, k3, v2, v1)
  243.     )
  244.     DELETE ON v2 >100
  245.     WITH BROKER my_hdfs_broker
  246.     (
  247.     "username" = "hdfs_user",
  248.     "password" = "hdfs_passwd"
  249.     )
  250.     PROPERTIES
  251.     (
  252.     "timeout" = "3600",
  253.     "max_filter_ratio" = "0.1"
  254.     );
  256. 13. Filter the original data first, and perform column mapping, conversion and filtering operations
  258.     LOAD LABEL example_db.label_filter
  259.     (
  260.      DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  261.      INTO TABLE `tbl1`
  262.      COLUMNS TERMINATED BY ","
  263.      (k1,k2,v1,v2)
  264.      PRECEDING FILTER k1 > 2
  265.      SET (k1 = k1 +1)
  266.      WHERE k1 > 3
  267.     ) 
  268.     with BROKER "hdfs" ("username"="user", "password"="pass");
  270. 14. Import the data in the json file, and specify format as json, it is judged by the file suffix by default, set parameters for reading data
  272.     LOAD LABEL example_db.label9
  273.     (
  274.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  275.     INTO TABLE `my_table`
  276.     FORMAT AS "json"
  277.     (k1, k2, k3)
  278.     properties("fuzzy_parse"="true", "strip_outer_array"="true")
  279.     )
  280.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");   
  282. 15. LOAD WITH HDFS, normal HDFS cluster
  283.     LOAD LABEL example_db.label_filter
  284.     (
  285.         DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  286.         INTO TABLE `tbl1`
  287.         COLUMNS TERMINATED BY ","
  288.         (k1,k2,v1,v2)
  289.     ) 
  290.     with HDFS (
  291.         "fs.defaultFS"="hdfs://testFs",
  292.         "hdfs_user"="user"
  293.     );
  294. 16. LOAD WITH HDFS, hdfs ha
  295.     LOAD LABEL example_db.label_filter
  296.     (
  297.         DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  298.         INTO TABLE `tbl1`
  299.         COLUMNS TERMINATED BY ","
  300.         (k1,k2,v1,v2)
  301.     ) 
  302.     with HDFS (
  303.         "fs.defaultFS"="hdfs://testFs",
  304.         "hdfs_user"="user"
  305.         "dfs.nameservices"="my_ha",
  306.         "dfs.ha.namenodes.xxx"="my_nn1,my_nn2",
  307.         "dfs.namenode.rpc-address.xxx.my_nn1"="host1:port",
  308.         "dfs.namenode.rpc-address.xxx.my_nn2"="host2:port",
  309.         "dfs.client.failover.proxy.provider.xxx"="org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  310.     );

keyword

  1. BROKER,LOAD