DataWorks:OSS データソース

オフライン読み取り

OSS Reader は、非構造化データを格納するためのサービスである OSS からデータを読み取り、そのデータを Data Integration プロトコルに変換します。OSS Reader は、次の機能をサポートしています。

オフライン書き込み

OSS Writer は、データ同期プロトコルからテキストファイルにデータを変換し、非構造化データを格納するためのサービスである 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 形式のファイルを読み取るために、DataWorks は 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 テキストファイルの最初の列から整数フィールドを読み取ります。
    },
    {
       "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":"",// Data Integration は、null と解釈できる文字列を定義するために nullFormat パラメーターを提供します。
                "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 スクリプトデモ: ORC または Parquet ファイルを OSS に書き込むスクリプトの設定

DataWorks は 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 データ型がサポートされています:

  フィールドタイプ	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"
  }
  }

• データを 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 {
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;
}