DataWorks:OSS データソース

オフライン読み取り

OSS Reader は OSS からデータを読み取り、Data Integration プロトコルに変換します。OSS は非構造化データ向けのストレージサービスです。Data Integration のために、OSS Reader は以下の機能をサポートしています。

オフライン書き込み

OSS Writer は、データ同期プロトコルから OSS 内のテキストファイルにデータを変換します。OSS は非構造化データ向けのストレージサービスです。OSS Writer は以下の機能をサポートしています。

リアルタイム書き込み

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

• Hudi (0.12.x)、Paimon、Iceberg などのデータレイクへの単一テーブルからのリアルタイム書き込みをサポートしています。

単一テーブルのオフライン同期タスクの設定ガイド

• 設定手順の詳細については、「コードレス UI 設定」および「コードエディタ設定」をご参照ください。

• コードエディタのすべてのパラメーターおよびスクリプトデモについては、「付録：スクリプトデモとパラメーター説明」をご参照ください。

単一テーブルのリアルタイム同期タスクの設定ガイド

構成プロセスの詳細については、Data Integration でリアルタイム同期タスクを設定するおよびDataStudio でリアルタイム同期タスクを設定するをご参照ください。

データベース全体の同期の設定ガイド

構成プロセスの詳細については、「データベース全体のオフライン同期タスク」および「データベース全体のリアルタイム同期タスク」をご参照ください。

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

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

Reader スクリプトデモ：一般的な例

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"oss",// プラグイン名。
            "parameter":{
                "nullFormat":"",// null を表す文字列を定義します。
                "compress":"",// テキスト圧縮タイプ。
                "datasource":"",// データソース。
                "column":[// フィールド。
                    {
                        "index":0,// カラムインデックス。
                        "type":"string"// データの型。
                    },
                    {
                        "index":1,
                        "type":"long"
                    },
                    {
                        "index":2,
                        "type":"double"
                    },
                    {
                        "index":3,
                        "type":"boolean"
                    },
                    {
                        "format":"yyyy-MM-dd HH:mm:ss", // 時間フォーマット。
                        "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,// throttle が false の場合、mbps パラメーターは無効になり、レートは制限されません。throttle が true の場合、レートは制限されます。
            "concurrent":1, // 同時実行ジョブ数。
            "mbps":"12"// レート制限。1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Reader スクリプトデモ：OSS から ORC または Parquet ファイルを読み取る

OSS から ORC または Parquet 形式のファイルを読み取るには、HDFS Reader を再利用できます。既存の 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 テキストファイルの最初のカラムから int フィールドを取得します。
    },
    {
       "type": "string",
       "value": "alibaba"  // OSS Reader 内で文字列フィールド "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", // 同期されたデータを単一の OSS ファイルに書き込むかどうかを指定します。
                "encoding":"",// エンコード形式。
                "fieldDelimiter":",",// 列区切り文字。
                "fileFormat":"",// テキストファイル形式。
                "object":""// オブジェクトプレフィックス。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラーレコード数。
        },
        "speed":{
            "throttle":true,// throttle が false の場合、mbps パラメーターは無効になり、レートは制限されません。throttle が true の場合、レートは制限されます。
            "concurrent":1, // 同時実行ジョブ数。
            "mbps":"12"// レート制限。1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

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

OSS への ORC または Parquet ファイルの書き込みには、HDFS Writer を再利用できます。既存の OSS Writer パラメーターに加えて、Path および FileFormat などの拡張パラメーターを使用できます。これらのパラメーターの詳細については、「HDFS Writer」をご参照ください。

以下は、OSS への ORC または Parquet ファイルの書き込み例です。

• OSS への ORC ファイルの書き込み

  ORC ファイルを書き込むには、コードエディタを使用する必要があります。fileFormat を orc に設定し、path を書き込むファイルのパスに設定し、column を {"name": "your column name", "type": "your column type"} の形式で設定します。

  書き込みをサポートしている ORC 型は以下のとおりです。

  フィールド型	OSS へのオフライン書き込み（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"
  }
  ],
  "writeMode": "append",
  "fieldDelimiter": "\t",
  "compress": "NONE",
  "encoding": "UTF-8"
  }
  }

• OSS への Parquet ファイルの書き込み

  {
  "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 {
Required/Optional, 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;
}