DataWorks:OSS

バッチデータの読み取り

OSS Reader は OSS からデータを読み取り、Data Integration が処理できる形式に変換します。OSS は非構造化データストレージサービスであるため、Reader は以下の特徴をサポートします。

バッチデータの書き込み

OSS Writer は Data Integration 準拠のデータを変換し、テキストファイルとして OSS に書き込みます。OSS は非構造化データストレージサービスであるため、Writer は以下の特徴をサポートします。

リアルタイムデータの書き込み

• リアルタイムデータ書き込みをサポートします。

• 単一テーブルから、Hudi (0.12.x)、Paimon、Iceberg などのデータレイク形式へのリアルタイム書き込みを実行できます。

単一テーブルのバッチ同期

• 「コードレス UI での設定」および「スクリプトモードでの設定」をご参照ください。

• スクリプトモードのパラメーターとコードサンプルについては、「付録：コードサンプルとパラメーターの説明」をご参照ください。

単一テーブルのリアルタイム同期

「単一テーブルのリアルタイム同期タスクの設定」をご参照ください。

データベース全体の同期

「データベース全体のバッチ同期タスクの設定」および「データベース全体のリアルタイム同期タスクの設定」をご参照ください。

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプト形式の要件に基づいて、スクリプト内の関連パラメーターを設定する必要があります。詳細については、「コードエディタの使用」をご参照ください。以下では、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明します。

一般的な Reader スクリプトの例

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"oss",// プラグイン名。
            "parameter":{
                "nullFormat":"",// null 値を表す文字列。
                "compress":"",// 圧縮タイプ。
                "datasource":"",// データソース。
                "column":[// 列の定義。
                    {
                        "index":0,// 列インデックス。0 から始まります。
                        "type":"string"// データの型。
                    },
                    {
                        "index":1,
                        "type":"long"
                    },
                    {
                        "index":2,
                        "type":"double"
                    },
                    {
                        "index":3,
                        "type":"boolean"
                    },
                    {
                        "format":"yyyy-MM-dd HH:mm:ss", // 時刻フォーマット。「type」が「date」の場合に必須です。
                        "index":4,
                        "type":"date"
                    }
                ],
                "skipHeader":"",// CSV 形式のファイルでヘッダーをスキップするかどうか。
                "encoding":"",// エンコーディング。
                "fieldDelimiter":",",// フィールドデリミタ。
                "fileFormat": "",// ファイル形式。
                "object":[]// 読み取るファイルのオブジェクトプレフィックス。
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":""// 許容されるエラーレコードの最大数。
        },
        "speed":{
            "throttle":true,// true の場合、速度制限を有効にします。false の場合、「mbps」パラメーターは無視されます。
            "concurrent":1, // ジョブの同時実行数。
            "mbps":"12"// 速度制限の上限 (MB/s)。(1 mbps = 1 MB/s)。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

OSS からの ORC または Parquet ファイルの読み取り

HDFS Reader を再利用することで、OSS から ORC または Parquet ファイルを読み取ることができます。そのためには、標準の OSS Reader パラメーターに加えて、path (ORC の場合) および fileFormat (ORC および Parquet の場合) パラメーターを指定します。

• この例は、OSS から ORC 形式のデータを読み取る方法を示しています。

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "orc",
  "path": "/tests/case61/orc__691b6815_9260_4037_9899_****",
  "column": [
  {
  "index": 0,
  "type": "long"
  },
  {
  "index": "1",
  "type": "string"
  },
  {
  "index": "2",
  "type": "string"
  }
  ]
  }
  }

• この例は、OSS から Parquet 形式のデータを読み取る方法を示しています。

  {
    "type":"job",
      "version":"2.0",
      "steps":[
      {
        "stepType":"oss",
        "parameter":{
          "nullFormat":"",
          "compress":"",
          "fileFormat":"parquet",
          "path":"/*",
          "parquetSchema":"message m { optional BINARY registration_dttm (UTF8); optional Int64 id; optional BINARY first_name (UTF8); optional BINARY last_name (UTF8); optional BINARY email (UTF8); optional BINARY gender (UTF8); optional BINARY ip_address (UTF8); optional BINARY cc (UTF8); optional BINARY country (UTF8); optional BINARY birthdate (UTF8); optional DOUBLE salary; optional BINARY title (UTF8); optional BINARY comments (UTF8); }",
          "column":[
            {
              "index":"0",
              "type":"string"
            },
            {
              "index":"1",
              "type":"long"
            },
            {
              "index":"2",
              "type":"string"
            },
            {
              "index":"3",
              "type":"string"
            },
            {
              "index":"4",
              "type":"string"
            },
            {
              "index":"5",
              "type":"string"
            },
            {
              "index":"6",
              "type":"string"
            },
            {
              "index":"7",
              "type":"string"
            },
            {
              "index":"8",
              "type":"string"
            },
            {
              "index":"9",
              "type":"string"
            },
            {
              "index":"10",
              "type":"double"
            },
            {
              "index":"11",
              "type":"string"
            },
            {
              "index":"12",
              "type":"string"
            }
          ],
          "skipHeader":"false",
          "encoding":"UTF-8",
          "fieldDelimiter":",",
          "fieldDelimiterOrigin":",",
          "datasource":"wpw_demotest_oss",
          "envType":0,
          "object":[
            "wpw_demo/userdata1.parquet"
          ]
        },
        "name":"Reader",
        "category":"reader"
      },
      {
        "stepType":"odps",
        "parameter":{
          "partition":"dt=${bizdate}",
          "truncate":true,
          "datasource":"0_odps_wpw_demotest",
          "envType":0,
          "column":[
            "id"
          ],
          "emptyAsNull":false,
          "table":"wpw_0827"
        },
        "name":"Writer",
        "category":"writer"
      }
    ],
      "setting":{
      "errorLimit":{
        "record":""
      },
      "locale":"zh_CN",
        "speed":{
        "throttle":false,
          "concurrent":2
      }
    },
    "order":{
      "hops":[
        {
          "from":"Reader",
          "to":"Writer"
        }
      ]
    }
  }

Reader スクリプトのパラメーター

message MessageTypeName {
required/optional, data_type, column_name;
......................;
}

"parquetSchema": "message m { optional int32 minute_id; optional int32 dsp_id; optional int32 adx_pid; optional int64 req; optional int64 res; optional int64 suc; optional int64 imp; optional double revenue; }"

"column": ["*"]

"column":
    {
       "type": "long",
       "index": 0    // OSS テキストファイルの最初の列から整数を読み取ります。
    },
    {
       "type": "string",
       "value": "alibaba"  // 現在の列に対して定数文字列「alibaba」を生成します。
    }

Writer スクリプトの例

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"oss",// プラグイン名。
            "parameter":{
                "nullFormat":"",// どの文字列を null 値として解釈するかを定義します。
                "dateFormat":"",// 日付フォーマット。
                "datasource":"",// データソース。
                "writeMode":"",// 書き込みモード。
                "writeSingleObject":"false", // true の場合、すべてのデータを単一の OSS ファイルに書き込みます。
                "encoding":"",// ファイルエンコーディング。
                "fieldDelimiter":",",// フィールドデリミタ。
                "fileFormat":"",// ファイル形式。
                "object":""// オブジェクトプレフィックス。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラー上限。
        },
        "speed":{
            "throttle":true,// 速度制限を有効にします。false の場合、「mbps」パラメーターは無視されます。
            "concurrent":1, // ジョブの同時実行数。
            "mbps":"12"// 速度制限の上限 (MB/s)。(1 mbps = 1 MB/s)。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

スクリプトサンプル：OSS への ORC または Parquet ファイルの書き込み

HDFS Writer を使用して、ORC または Parquet ファイルを OSS に書き込むことができます。このメソッドは、既存の OSS Writer パラメーターに加えて、path や fileFormat などの拡張設定パラメーターを追加します。これらのパラメーターの説明については、「HDFS Writer」をご参照ください。

以下の例は、ORC または Parquet ファイルを OSS に書き込む方法を示しています：

• ORC ファイルを OSS に書き込む

  現在、ORC ファイルはコードエディタでのみ書き込み可能です。タスクを設定するには、コードエディタに切り替える必要があります。fileFormat パラメーターを orc に設定し、ファイルの path を指定し、各 column を {"name":"your column name","type": "your column type"} の形式で定義します。

  オフライン書き込みでサポートされている ORC の列の型は以下の通りです：

  型	ステータス
  TINYINT	サポート対象
  SMALLINT	サポート対象
  INT	サポート対象
  BIGINT	サポート対象
  FLOAT	サポート対象
  DOUBLE	サポート対象
  TIMESTAMP	サポート対象
  DATE	サポート対象
  VARCHAR	サポート対象
  STRING	サポート対象
  CHAR	サポート対象
  BOOLEAN	サポート対象
  DECIMAL	サポート対象
  BINARY	サポート対象

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "orc",
  "path": "/tests/case61",
  "fileName": "orc",
  "writeMode": "append",
  "column": [
  {
  "name": "col1",
  "type": "BIGINT"
  },
  {
  "name": "col2",
  "type": "DOUBLE"
  },
  {
  "name": "col3",
  "type": "STRING"
  }
  ],
  "fieldDelimiter": "\t",
  "compress": "NONE",
  "encoding": "UTF-8"
  }
  }

• Parquet ファイルを OSS に書き込む

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "parquet",
  "path": "/tests/case61",
  "fileName": "test",
  "writeMode": "append",
  "fieldDelimiter": "\t",
  "compress": "SNAPPY",
  "encoding": "UTF-8",
  "parquetSchema": "message test { required int64 int64_col;\n required binary str_col (UTF8);\nrequired group params (MAP) {\nrepeated group key_value {\nrequired binary key (UTF8);\nrequired binary value (UTF8);\n}\n}\nrequired group params_arr (LIST) {\nrepeated group list {\nrequired binary element (UTF8);\n}\n}\nrequired group params_struct {\nrequired int64 id;\n required binary name (UTF8);\n }\nrequired group params_arr_complex (LIST) {\nrepeated group list {\nrequired group element {\n required int64 id;\n required binary name (UTF8);\n}\n}\n}\nrequired group params_complex (MAP) {\nrepeated group key_value {\nrequired binary key (UTF8);\nrequired group value {\nrequired int64 id;\n required binary name (UTF8);\n}\n}\n}\nrequired group params_struct_complex {\nrequired int64 id;\n required group detail {\nrequired int64 id;\n required binary name (UTF8);\n}\n}\n}",
  "dataxParquetMode": "fields"
  }
  }

Writer スクリプトのパラメーター

message MessageTypeName {
modifier, data_type, column_name;
......................;
}

message m {
optional int64 id;
optional int64 date_id;
optional binary datetimestring;
optional int32 dspId;
optional int32 advertiserId;
optional int32 status;
optional int64 bidding_req_num;
optional int64 imp;
optional int64 click_num;
}