datax odpswriter 使用

2022-07-14 浏览 (2507)

DataX ODPS写入

1 快速介绍

ODPSWriter插件用于实现往ODPS插入或者更新数据，主要提供给etl开发同学将业务数据导入odps，适合于TB,GB数量级的数据传输，如果需要传输PB量级的数据，请选择dt task工具 ;

2 实现原理

在底层实现上，ODPSWriter是通过DT Tunnel写入ODPS系统的，有关ODPS的更多技术细节请参看 ODPS主站 https://data.aliyun.com/product/odps 和ODPS产品文档 https://help.aliyun.com/product/27797.html

目前 DataX3 依赖的 SDK 版本是：

                <dependency>
                    <groupId>com.aliyun.odps</groupId>
                    <artifactId>odps-sdk-core-internal</artifactId>
                    <version>0.13.2</version>
                </dependency>

注意: 如果你需要使用ODPSReader/Writer插件，请务必使用JDK 1.6-32及以上版本 使用java -version查看Java版本号

3 功能说明

3.1 配置样例

  • 这里使用一份从内存产生到ODPS导入的数据。
{
  "job": {
    "setting": {
      "speed": {
        "byte": 1048576
      }
    },
    "content": [
      {
        "reader": {
          "name": "streamreader",
          "parameter": {
            "column": [
              {
                "value": "DataX",
                "type": "string"
              },
              {
                "value": "test",
                "type": "bytes"
              }
            ],
            "sliceRecordCount": 100000
          }
        },
        "writer": {
          "name": "odpswriter",
          "parameter": {
            "project": "chinan_test",
            "table": "odps_write_test00_partitioned",
            "partition": "school=SiChuan-School,class=1",
            "column": [
              "id",
              "name"
            ],
            "accessId": "xxx",
            "accessKey": "xxxx",
            "truncate": true,
            "odpsServer": "http://sxxx/api",
            "tunnelServer": "http://xxx",
            "accountType": "aliyun"
          }
        }
      }
    ]
  }
}

3.2 参数说明

  • accessId

    • 描述：ODPS系统登录ID
    • 必选：是
    • 默认值：无

  • accessKey

    • 描述：ODPS系统登录Key
    • 必选：是
    • 默认值：无

  • project

    • 描述：ODPS表所属的project，注意:Project只能是字母+数字组合，请填写英文名称。在云端等用户看到的ODPS项目中文名只是显示名，请务必填写底层真实地Project英文标识名。
    • 必选：是
    • 默认值：无

  • table

    • 描述：写入数据的表名，不能填写多张表，因为DataX不支持同时导入多张表。
    • 必选：是
    • 默认值：无

  • partition

    • 描述：需要写入数据表的分区信息，必须指定到最后一级分区。把数据写入一个三级分区表，必须配置到最后一级分区，例如pt=20150101/type＝1/biz=2。
    • 必选：如果是分区表，该选项必填，如果非分区表，该选项不可填写。
    • 默认值：空

  • column

    • 描述：需要导入的字段列表，当导入全部字段时，可以配置为"column": ["*"], 当需要插入部分odps列填写部分列，例如"column": ["id", "name"]。ODPSWriter支持列筛选、列换序，例如表有a,b,c三个字段，用户只同步c,b两个字段。可以配置成["c","b"], 在导入过程中，字段a自动补空，设置为null。
    • 必选：否
    • 默认值：无

  • truncate

    • 描述：ODPSWriter通过配置"truncate": true，保证写入的幂等性，即当出现写入失败再次运行时，ODPSWriter将清理前述数据，并导入新数据，这样可以保证每次重跑之后的数据都保持一致。

      **truncate选项不是原子操作！ODPS SQL无法做到原子性。因此当多个任务同时向一个Table/Partition清理分区时候，可能出现并发时序问题，请务必注意！**针对这类问题，我们建议尽量不要多个作业DDL同时操作同一份分区，或者在多个并发作业启动前，提前创建分区。

    • 必选：是

    • 默认值：无

  • odpsServer

    • 描述：ODPS的server地址，线上地址为 http://service.odps.aliyun.com/api
    • 必选：是
    • 默认值：无

  • tunnelServer

    • 描述：ODPS的tunnelserver地址，线上地址为 http://dt.odps.aliyun.com
    • 必选：是，
    • 默认值：无

3.3 类型转换

类似ODPSReader，目前ODPSWriter支持大部分ODPS类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。

下面列出ODPSWriter针对ODPS类型转换列表:

DataX 内部类型ODPS 数据类型
Longbigint
Doubledouble
Stringstring
Datedatetime
Booleanbool

4 插件特点

4.1 关于列筛选的问题

  • ODPS本身不支持列筛选、重排序、补空等等，但是DataX ODPSWriter完成了上述需求，支持列筛选、重排序、补空。例如需要导入的字段列表，当导入全部字段时，可以配置为"column": ["*"]，odps表有a,b,c三个字段，用户只同步c,b两个字段，在列配置中可以写成"column": ["c","b"]，表示会把reader的第一列和第二列导入odps的c字段和b字段，而odps表中新插入记录的a字段会被置为null.

4.2 列配置错误的处理

  • 为了保证写入数据的可靠性，避免多余列数据丢失造成数据质量故障。对于写入多余的列，ODPSWriter将报错。例如ODPS表字段为a,b,c，但是ODPSWriter写入的字段为多于3列的话ODPSWriter将报错。

4.3 分区配置注意事项

  • ODPSWriter只提供 写入到最后一级分区 功能，不支持写入按照某个字段进行分区路由等功能。假设表一共有3级分区，那么在分区配置中就必须指明写入到某个三级分区，例如把数据写入一个表的第三级分区，可以配置为 pt=20150101/type＝1/biz=2，但是不能配置为pt=20150101/type＝1或者pt=20150101。

4.4 任务重跑和failover

  • ODPSWriter通过配置"truncate": true，保证写入的幂等性，即当出现写入失败再次运行时，ODPSWriter将清理前述数据，并导入新数据，这样可以保证每次重跑之后的数据都保持一致。如果在运行过程中因为其他的异常导致了任务中断，是不能保证数据的原子性的，数据不会回滚也不会自动重跑，需要用户利用幂等性这一特点重跑去确保保证数据的完整性。truncate为true的情况下，会将指定分区\表的数据全部清理，请谨慎使用！

5 性能报告（线上环境实测）

5.1 环境准备

5.1.1 数据特征

建表语句：

use cdo_datasync;
create table datax3_odpswriter_perf_10column_1kb_00(
s_0 string,
bool_1 boolean,
bi_2 bigint,
dt_3 datetime,
db_4 double,
s_5 string,
s_6 string,
s_7 string,
s_8 string,
s_9 string
)PARTITIONED by (pt string,year string);

单行记录类似于：

s_0    : 485924f6ab7f272af361cd3f7f2d23e0d764942351#$%^&fdafdasfdas%%^(*&^^&*
bool_1 : true
bi_2   : 1696248667889
dt_3   : 2013-07-0600: 00: 00
db_4   : 3.141592653578
s_5    : 100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11209
s_6    : 100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11fdsafdsfdsa209
s_7    : 100DAFDSAFDSAHOFJDPSAWIFDISHAF;dsadsafdsahfdsajf;dsfdsa;FJDSAL;11209
s_8    : 100dafdsafdsahofjdpsawifdishaf;DSADSAFDSAHFDSAJF;dsfdsa;fjdsal;11209
s_9    : 12~!2345100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11209

5.1.2 机器参数

  • 执行DataX的机器参数为:

    1. cpu : 24 Core Intel(R) Xeon(R) CPU E5-2630 0 @ 2.30GHz cache 15.36MB
    2. mem : 50GB
    3. net : 千兆双网卡
    4. jvm : -Xms1024m -Xmx1024m -XX:+HeapDumpOnOutOfMemoryError
    5. disc: DataX 数据不落磁盘，不统计此项

  • 任务配置为:

{
    "job": {
        "setting": {
            "speed": {
                "channel": "1,2,4,5,6,8,16,32,64"
            }
        },
        "content": [
            {
                "reader": {
                    "name": "odpsreader",
                    "parameter": {
                        "accessId": "******************************",
                        "accessKey": "*****************************",
                        "column": [
                            "*"
                        ],
                        "partition": [
                            "pt=20141010000000,year=2014"
                        ],
                        "odpsServer": "http://service.odps.aliyun.com/api",
                        "project": "cdo_datasync",
                        "table": "datax3_odpswriter_perf_10column_1kb_00",
                        "tunnelServer": "http://dt.odps.aliyun.com"
                    }
                },
                "writer": {
                    "name": "streamwriter",
                    "parameter": {
                        "print": false,
                        "column": [
                            {
                                "value": "485924f6ab7f272af361cd3f7f2d23e0d764942351#$%^&fdafdasfdas%%^(*&^^&*"
                            },
                            {
                                "value": "true",
                                "type": "bool"
                            },
                            {
                                "value": "1696248667889",
                                "type": "long"
                            },
                            {
                                "type": "date",
                                "value": "2013-07-06 00:00:00",
                                "dateFormat": "yyyy-mm-dd hh:mm:ss"
                            },
                            {
                                "value": "3.141592653578",
                                "type": "double"
                            },
                            {
                                "value": "100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11209"
                            },
                            {
                                "value": "100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11fdsafdsfdsa209"
                            },
                            {
                                "value": "100DAFDSAFDSAHOFJDPSAWIFDISHAF;dsadsafdsahfdsajf;dsfdsa;FJDSAL;11209"
                            },
                            {
                                "value": "100dafdsafdsahofjdpsawifdishaf;DSADSAFDSAHFDSAJF;dsfdsa;fjdsal;11209"
                            },
                            {
                                "value": "12~!2345100dafdsafdsahofjdpsawifdishaf;dsadsafdsahfdsajf;dsfdsa;fjdsal;11209"
                            }
                        ]
                    }
                }
            }
        ]
    }
}

5.2 测试报告

并发任务数blockSizeInMBDataX速度(Rec/s)DataX流量(MB/S)网卡流量(MB/S)DataX运行负载
1323030313.0314.50.12
1643846116.5416.50.44
11284645420.5526.70.47
12565263122.6426.70.47
15125882325.3028.70.44
43211481649.3855.30.75
46414757763.4771.30.82
412817774476.4583.20.97
425617391374.8080.11.01
451220000086.0295.11.41
83220448087.9592.71.16
864294224126.55135.31.65
8128365475157.19163.72.89
8256394713169.83176.72.72
8512241691103.95125.72.29
1632420838181.01198.02.56
1664458144197.05217.42.85
16128443219190.63210.53.29
16256315235135.58140.00.95
16512OOM

说明：

  1. OdpsWriter 影响速度的是channel 和 blockSizeInMB。blockSizeInMB 取32 和 64时，速度比较稳定,过分大的 blockSizeInMB 可能造成速度波动以及内存OOM。
  2. channel 和 blockSizeInMB 对速度的影响都很明显，建议综合考虑配合选择。
  3. channel 数目的选择，可以综合考虑源端数据特征进行选择，对于StreamReader，在16个channel时将网卡打满。

6 FAQ

1 导数据到 odps 的日志中有以下报错，该怎么处理呢？"ODPS-0420095: Access Denied - Authorization Failed [4002], You doesn‘t exist in project example_dev“

解决办法 :找ODPS Prject 的 owner给用户的云账号授权，授权语句： grant Describe,Select,Alter,Update on table [tableName] to user XXX

2 可以导入数据到odps的视图吗？

目前不支持通过视图到数据到odps,视图是ODPS非实体化数据存储对象，技术上无法向视图导入数据。

相关文章

  • 所属分类： 大数据
  • 本文标签： 大数据
  • 版权声明： 本文链接 https://seaxiang.com/blog/b55049e842e24105b8933fcd069a879d