Broker Load

Broker Load导入

查看Broker

SHOW PROC "/brokers";

创建导入任务

• 语法

  LOAD LABEL [<database_name>.]<label_name>
  (
      data_desc[, data_desc ...]
  )
  WITH BROKER
  [broker_properties]
  [opt_properties]

• 参数描述

  • Label

    导入任务的标识。每个导入任务都有一个唯一的Label，由系统自动生成或在导入命令中自定义。Label可以用来防止导入相同的数据，并查看对应导入任务的执行情况。当Label对应的导入任务状态为FINISHED时，对应的Label无法再次使用；当状态为CANCELLED时，可以再次使用该Label提交导入作业。

  • 数据描述类data_desc

    data_desc:
        DATA INFILE ('file_path', ...)
        [NEGATIVE]
        INTO TABLE tbl_name
        [PARTITION (p1, p2)]
        [COLUMNS TERMINATED BY column_separator ]
        [FORMAT AS file_type]
        [(col1, ...)]
        [COLUMNS FROM PATH AS (colx, ...)]
        [SET (k1=f1(xx), k2=f2(xx))]
        [WHERE predicate]

    相关参数描述如下表所示。

    参数	描述
    file_path	文件路径可以指定到文件，也可以用星号（*）通配符指定某个目录下的所有文件。中间的目录也可以使用通配符匹配。 可以使用的通配符有? * [] {} ^，使用规则请参见FileSystem。 例如， 通过oss://bucket/data/tablename , 可以匹配tablename下所有分区内的所有文件。通过oss://bucket/data/tablename/dt=202301, 可以匹配tablename下1月分区的所有文件。
    negative	设置数据取反导入。 该功能适用的场景是当数据表中聚合列的类型均为SUM类型时，如果希望撤销某一批导入的数据，可以通过negative参数导入同一批数据，StarRocks会自动为这批数据在聚合列上数据取反，以达到消除同一批数据的功能。
    partition	指定待导入表的Partition信息。 如果待导入数据不属于指定的Partition，则不会被导入。同时，不指定Partition中的数据会被认为是“错误数据”。对于不想导入，也不想记录为“错误数据”的数据，可以使用where predicate来过滤。
    column_separator	COLUMNS TERMINATED BY column_separator，用于指定导入文件中的列分隔符，默认为\t。 如果是不可见字符，则需要加\x作为前缀，使用十六进制来表示分隔符。例如，Hive文件的分隔符为\x01，则列分隔符为\\x01。
    file_type	FORMAT AS file_type，用于指定导入文件的类型。例如，parquet、orc、csv，默认值为csv。 parquet类型也可以通过文件后缀名.parquet或者.parq判断。
    COLUMNS FROM PATH AS	提取文件路径中的分区字段。 例如，导入文件为/path/col_name=col_value/dt=20210101/file1，其中col_name/dt为表中的列，则将col_value、20210101分别导入到col_name和dt对应的列的代码示例如下。 (col1, col2) COLUMNS FROM PATH AS (col_name, dt)
    set column mapping	SET (k1=f1(xx), k2=f2(xx))，data_desc中的SET语句负责设置列函数变换。 如果原始数据的列和表中的列不一一对应，则需要使用该属性。
    where predicate	WHERE predicate，data_desc中的WHERE语句负责过滤已经完成transform的数据。 被过滤的数据不会进入容忍率的统计中。如果多个data_desc中声明了关于同一张表的多个条件，则会以AND语义合并这些条件。

  • 导入作业参数

    导入作业参数是指Broker Load创建导入语句中属于broker_properties部分的参数。导入作业参数是作用于整个导入作业的。

    broker_properties:
        (key2=value2, ...)

    部分参数描述如下表所示。

    参数	描述
    timeout	导入作业的超时时间（以秒为单位）。 您可以在opt_properties中自行设置每个导入的超时时间。导入任务在设定的时限内未完成则会被系统取消，变为CANCELLED。Broker Load的默认导入超时时间为4小时。 重要 通常情况下，不需要您手动设置导入任务的超时时间。当在默认超时时间内无法完成导入时，可以手动设置任务的超时时间。 推荐超时时间的计算方式为：超时时间 >（（总文件大小 （MB）* 待导入的表及相关Roll up表的个数） / (30 * 导入并发数）) 公式中的30为目前BE导入的平均速度，表示30 MB/s。例如，如果待导入数据文件为1 GB，待导入表包含2个Rollup表，当前的导入并发数为3，则timeout的最小值为 (1 * 1024 * 3 ) / (10 * 3) = 102 秒。 由于每个StarRocks集群的机器环境不同且集群并发的查询任务也不同，所以StarRocks集群的最慢导入速度需要您根据历史的导入任务速度进行推测。
    max_filter_ratio	导入任务的最大容忍率，默认为0容忍，取值范围是0~1。当导入的错误率超过该值，则导入失败。如果您希望忽略错误的行，可以设置该参数值大于0，来保证导入可以成功。 计算公式为：max_filter_ratio = (dpp.abnorm.ALL / (dpp.abnorm.ALL + dpp.norm.ALL ) ) 其中，dpp.abnorm.ALL表示数据质量不合格的行数，例如类型不匹配、列数不匹配和长度不匹配等。dpp.abnorm.ALL指的是导入过程中正确数据的条数，可以通过SHOW LOAD命令查询导入任务的正确数据量。 原始文件的行数 = dpp.abnorm.ALL + dpp.norm.ALL
    load_mem_limit	导入内存限制。默认值为0，表示不限制。
    strict_mode	Broker Load导入可以开启Strict Mode模式。开启方式为properties ("strict_mode" = "true") 。 默认关闭。 Strict Mode模式是对于导入过程中的列类型转换进行严格过滤。严格过滤的策略为，对于列类型转换，如果Strict Mode为true，则错误的数据将被过滤掉。错误数据是指原始数据并不为空值，在参与列类型转换后结果为空值的数据。但以下场景除外： 对于导入的某列由函数变换生成时，Strict Mode对其不产生影响。 对于导入的某列类型包含范围限制的，如果原始数据能正常通过类型转换，但无法通过范围限制的，Strict Mode对其也不产生影响。例如，如果类型是decimal(1,0)，原始数据为10，则属于可以通过类型转换但不在列声明的范围内，Strict Mode对其不产生影响。

查看导入任务状态

SHOW LOAD;

{"Unfinished backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"ScannedRows":2390016,"TaskNumber":1,"All backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"FileNumber":1,"FileSize":1073741824}

取消导入任务

当Broker Load作业状态不为CANCELLED或FINISHED时，可以手动取消。取消时需要指定待取消导入任务的Label。

导入任务并发度

实例总数的计算公式为实例的总数 = min（导入文件总大小/单个实例处理的最小数据量，单个任务最大并发实例数，单个BE上并发实例数 * BE数）。

通常情况下，一个作业只有一个DataDescription，只会拆分成一个任务。任务会拆成与BE数相等的实例，然后分配到所有BE上并行执行。

导入示例

阿里云OSS数据导入

1. 创建测试表。

   create database if not exists load_test;
   use load_test;
   create table if not exists customer(
     c_customer_sk bigint,
     c_customer_id char(16),
     c_current_cdemo_sk bigint,
     c_current_hdemo_sk bigint,
     c_current_addr_sk bigint,
     c_first_shipto_date_sk bigint,
     c_first_sales_date_sk bigint,
     c_salutation char(10),
     c_first_name char(20),
     c_last_name char(30),
     c_preferred_cust_flag char(1),
     c_birth_day int,
     c_birth_month int,
     c_birth_year int,
     c_birth_country varchar(20),
     c_login char(13),
     c_email_address char(50),
     c_last_review_date_sk bigint
   )
   duplicate key (c_customer_sk)
   distributed by hash(c_customer_sk) buckets 5
   properties(
     "replication_num"="1"
   );

2. 创建导入任务。

   下载customer.orc。

   LOAD LABEL load_test.customer_label
   (
     DATA INFILE("oss://{bucket_name}/data/customer.orc")
     INTO TABLE customer
     format as "orc"
   )
   WITH BROKER 'broker'
   (
     "fs.oss.accessKeyId" = "xxxxx",
     "fs.oss.accessKeySecret" = "xxxxx",
     "fs.oss.endpoint" = "oss-cn-xxx-internal.aliyuncs.com"
   );

3. 查看导入任务状态。

   show load where label='customer_label';

4. 查询表信息。

   • 示例1

     select count(1) from customer;

     返回信息如下。

     +----------+
     | count(1) |
     +----------+
     |  6000000 |
     +----------+
     1 row in set (0.10 sec)

   • 示例2

     select * from customer limit 2;

     返回信息如下。

     +---------------+------------------+--------------------+--------------------+-------------------+------------------------+-----------------------+--------------+--------------+-------------+-----------------------+-------------+---------------+--------------+-----------------+---------+------------------------+-----------------------+
     | c_customer_sk | c_customer_id    | c_current_cdemo_sk | c_current_hdemo_sk | c_current_addr_sk | c_first_shipto_date_sk | c_first_sales_date_sk | c_salutation | c_first_name | c_last_name | c_preferred_cust_flag | c_birth_day | c_birth_month | c_birth_year | c_birth_country | c_login | c_email_address        | c_last_review_date_sk |
     +---------------+------------------+--------------------+--------------------+-------------------+------------------------+-----------------------+--------------+--------------+-------------+-----------------------+-------------+---------------+--------------+-----------------+---------+------------------------+-----------------------+
     |             2 | AAAAAAAACAAAAAAA |             819667 |               1461 |            681655 |                2452318 |               2452288 | Dr.          | Amy          | Moses       | Y                     |           9 |             4 |         1966 | TOGO            | NULL    | Amy.M****@Ovk9KjHH.com |               2452318 |
     |             2 | AAAAAAAACAAAAAAA |             819667 |               1461 |            681655 |                2452318 |               2452288 | Dr.          | Amy          | Moses       | Y                     |           9 |             4 |         1966 | TOGO            | NULL    | Amy.M****@Ovk9KjHH.com |               2452318 |
     +---------------+------------------+--------------------+--------------------+-------------------+------------------------+-----------------------+--------------+--------------+-------------+-----------------------+-------------+---------------+--------------+-----------------+---------+------------------------+-----------------------+

HDFS导入

• HDFS导入语法示例

  LOAD LABEL db1.label1
  (
      DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file1")
      INTO TABLE tbl1
      COLUMNS TERMINATED BY ","
      (tmp_c1, tmp_c2)
      SET
      (
          id=tmp_c2,
          name=tmp_c1
      ),
  
      DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file2")
      INTO TABLE tbl2
      COLUMNS TERMINATED BY ","
      (col1, col2)
      where col1 > 1
  )
  WITH BROKER 'broker'
  (
      "username" = "hdfs_username",
      "password" = "hdfs_password"
  )
  PROPERTIES
  (
      "timeout" = "3600"
  );

  说明

  如果您的集群类型是新版数据湖（DataLake）、实时数据流（DataFlow）、数据分析（OLAP）、数据服务（DataServing）和自定义场景的集群， 请替换emr-header-1为master-1-1。

• HDFS认证

  社区版本的HDFS支持简单认证和Kerberos认证两种认证方式。

  • 简单认证（Simple）：用户的身份由与HDFS建立链接的客户端操作系统决定。涉及参数如下表。

    参数	描述
    hadoop.security.authentication	认证方式。默认值为simple。
    username	HDFS的用户名。
    password	HDFS的密码。

  • Kerberos认证：客户端的身份由用户自己的Kerberos证书决定。

    涉及参数如下表。

    参数	描述
    hadoop.security.authentication	认证方式。默认值为kerberos。
    kerberos_principal	指定Kerberos的Principal。
    kerberos_keytab	指定Kerberos的keytab文件路径。该文件必须为Broker进程所在服务器上的文件。
    kerberos_keytab_content	指定Kerberos中keytab文件内容经过Base64编码之后的内容。 重要 该参数和kerberos_keytab参数只需配置一个。

• HDFS HA配置

  通过配置NameNode HA，可以在NameNode切换时，自动识别到新的NameNode。配置以下参数用于访问以HA模式部署的HDFS集群。

  参数	描述
  dfs.nameservices	指定HDFS服务的名称，您可以自定义。 例如，设置dfs.nameservices为my_ha。
  dfs.ha.namenodes.xxx	自定义NameNode的名称，多个名称时以逗号（,）分隔。其中xxx为dfs.nameservices中自定义的名称。 例如，设置dfs.ha.namenodes.my_ha为my_nn。
  dfs.namenode.rpc-address.xxx.nn	指定NameNode的RPC地址信息。其中nn表示dfs.ha.namenodes.xxx中配置的NameNode的名称。 例如，设置dfs.namenode.rpc-address.my_ha.my_nn参数值的格式为host:port。
  dfs.client.failover.proxy.provider	指定Client连接NameNode的Provider，默认值为org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider。