DataWorks:OSS

Baca data batch

OSS Reader membaca data dari OSS dan mengonversinya ke format yang dapat diproses oleh Data Integration. Karena OSS merupakan layanan penyimpanan data tidak terstruktur, reader mendukung fitur-fitur berikut.

• Saat menyiapkan data di OSS, file CSV harus sesuai dengan format CSV standar. Misalnya, tanda kutip ganda (") dalam kolom harus di-escape dengan menggantinya menjadi dua tanda kutip ganda (""), karena jika tidak, file tersebut mungkin terpisah secara salah. Jika file berisi beberapa pemisah, disarankan untuk menggunakan tipe teks.

• OSS adalah sumber data tidak terstruktur yang menyimpan data dalam bentuk file. Sebelum melakukan sinkronisasi data, pastikan struktur bidang data yang akan disinkronkan sesuai dengan ekspektasi Anda. Jika struktur data sumber berubah, Anda harus mengonfirmasi ulang struktur bidang dalam konfigurasi Tugas. Jika tidak, kesalahan sinkronisasi data dapat terjadi.

Tulis data batch

OSS Writer mengonversi data dari protokol Data Integration menjadi file teks di OSS. Karena OSS merupakan layanan penyimpanan data tidak terstruktur, writer mendukung fitur-fitur berikut.

Tulis real-time

• Mendukung kemampuan penulisan real-time.

• Penulisan real-time single-table ke data lake didukung: Hudi (0.12.x), Paimon, dan Iceberg.

Panduan konfigurasi tugas sinkronisasi batch single-table

• Untuk prosedurnya, lihat Konfigurasi Antarmuka tanpa kode dan Konfigurasi mode skrip.

• Untuk semua parameter dan demo skrip yang tersedia dalam mode skrip, lihat Lampiran: demo skrip dan deskripsi parameter di bawah ini.

Panduan konfigurasi tugas sinkronisasi real-time single-table

Untuk prosedurnya, lihat Konfigurasi tugas sinkronisasi real-time single-table.

Panduan konfigurasi sinkronisasi seluruh database

Untuk prosedurnya, lihat Kemampuan Tugas Sinkronisasi Batch Seluruh Database dan Kemampuan Tugas Sinkronisasi Real-time Seluruh Database.

Konfigurasi tugas sinkronisasi batch menggunakan editor kode

Jika Anda ingin mengonfigurasi Tugas sinkronisasi batch menggunakan editor kode, Anda harus mengonfigurasi parameter terkait dalam skrip sesuai dengan persyaratan format skrip terpadu. Untuk informasi lebih lanjut, lihat Konfigurasi mode skrip. Informasi berikut menjelaskan parameter yang harus Anda konfigurasi untuk sumber data saat mengonfigurasi Tugas sinkronisasi batch menggunakan editor kode.

Demo skrip Reader: Contoh umum

Contoh berikut menunjukkan skrip umum untuk OSS Reader.

{
    "type":"job",
    "version":"2.0",//Nomor versi.
    "steps":[
        {
            "stepType":"oss",//Nama plug-in.
            "parameter":{
                "nullFormat":"",//String yang merepresentasikan nilai null.
                "compress":"",//Jenis kompresi teks.
                "datasource":"",//Sumber data.
                "column":[//Kolom.
                    {
                        "index":0,//Indeks kolom.
                        "type":"string"//Tipe data.
                    },
                    {
                        "index":1,
                        "type":"long"
                    },
                    {
                        "index":2,
                        "type":"double"
                    },
                    {
                        "index":3,
                        "type":"boolean"
                    },
                    {
                        "format":"yyyy-MM-dd HH:mm:ss", //Format waktu.
                        "index":4,
                        "type":"date"
                    }
                ],
                "skipHeader":"",//File mirip CSV mungkin berisi baris header. Atur parameter ini untuk melewati header.
                "encoding":"",//Format encoding.
                "fieldDelimiter":",",//Pemisah kolom.
                "fileFormat": "",//Tipe teks.
                "object":[]//Awalan objek.
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":""//Jumlah maksimum catatan error yang diizinkan.
        },
        "speed":{
            "throttle":true,//Jika throttle diatur ke false, parameter mbps tidak berlaku, yang berarti pembatasan kecepatan dinonaktifkan. Jika throttle diatur ke true, pembatasan kecepatan diaktifkan.
            "concurrent":1 //Jumlah thread konkuren.
            "mbps":"12",//Laju pembatasan. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Demo skrip Reader: Membaca file ORC atau Parquet dari OSS

OSS Reader membaca file ORC atau Parquet dari OSS dengan menggunakan kembali HDFS Reader. Selain parameter OSS Reader yang sudah ada, parameter tambahan berikut ditambahkan: Path (ORC) dan FileFormat (ORC, Parquet) .

• Contoh berikut menunjukkan cara membaca file ORC dari OSS.

  {
  "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"
  }
  ]
  }
  }

• Contoh berikut menunjukkan cara membaca file Parquet dari OSS.

  {
    "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"
        }
      ]
    }
  }

Demo skrip Reader: Membaca file JSONLine dari OSS, mengurai data JSON, dan menulis data ke tujuan

Contoh berikut menunjukkan cara membaca data JSONLine dari OSS, mengurai data JSON, melakukan vektorisasi pada bidang tertentu dalam data JSON, lalu menulis data ke Milvus:

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "oss",
            "parameter": {
                "datasource": "${OSS data source name}",
                "column": [
                    {
                        "name": "chunk_text",
                        "index": 0,
                        "type": "string"
                    }
                ],
                "fieldDelimiter": ",",
                "encoding": "UTF-8",
                "fileFormat": "jsonl",
                "object": [
                    "embedding/chunk1.jsonl"
                ]
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "json-extracting",
            "parameter": {
                "column": [
                    {
                        "name": "text",
                        "fromColumn": "chunk_text",
                        "jsonPath": "$.text",
                        "type": "STRING",
                        "nullOrInvalidDataAction": "DIRTY_DATA"
                    }
                ]
            },
            "name": "jsonextract",
            "category": "flatmap"
        },
        {
            "stepType": "embedding-transformer",
            "parameter": {
                "modelProvider": "bailian",
                "modelName": "text-embedding-v4",
                "apiKey": "${the AccessKey that you want to use}",
                "embeddingColumns": {
                    "sourceColumnNames": [
                        "text"
                    ],
                    "embeddingColumnName": "my_vector"
                },
                "batchSize": 8,
                "dimension": 1024
            },
            "name": "embedding",
            "category": "flatmap"
        },
        {
            "stepType": "milvus",
            "parameter": {
                "schemaCreateMode": "ignore",
                "enableDynamicSchema": true,
                "datasource": "${Milvus data source name}",
                "column": [
                    {
                        "name": "my_vector",
                        "type": "FloatVector",
                        "dimension": "1024",
                        "elementType": "None",
                        "maxLength": "65535"
                    },
                    {
                        "name": "text",
                        "type": "VarChar",
                        "elementType": "None",
                        "maxLength": "65535"
                    }
                ],
                "collection": "yunshi_vector_07171130",
                "writeMode": "insert",
                "batchSize": 1024,
                "columnMapping": [
                    {
                        "sourceColName": "my_vector",
                        "dstColName": "my_vector"
                    },
                    {
                        "sourceColName": "text",
                        "dstColName": "text"
                    }
                ]
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 1
        }
    }
}

Parameter skrip Reader

message MessageTypeName {
required_or_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    //Ambil bidang INT dari kolom pertama teks OSS.
    },
    {
       "type": "string",
       "value": "alibaba"  //Hasilkan bidang string "alibaba" dari OSSReader sebagai bidang saat ini.
    }

Demo skrip Writer: Contoh umum

Berikut adalah contoh skrip umum untuk OSS Writer.

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"oss",// Nama plug-in.
            "parameter":{
                "nullFormat":"",// Format null yang disediakan oleh Data Integration, yang mendefinisikan string apa yang dapat merepresentasikan null.
                "dateFormat":"",// Format tanggal.
                "datasource":"",// Sumber data.
                "writeMode":"",// Mode tulis.
                "writeSingleObject":"false", // Menentukan apakah data yang disinkronkan ditulis ke satu file OSS.
                "encoding":"",// Format encoding.
                "fieldDelimiter":","// Pemisah kolom.
                "fileFormat":"",// Format file.
                "object":""// Awalan objek.
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// Jumlah maksimum catatan error yang diizinkan.
        },
        "speed":{
            "throttle":true,// Jika throttle diatur ke false, parameter mbps tidak berlaku dan bandwidth tidak dibatasi. Jika throttle diatur ke true, bandwidth dibatasi.
            "concurrent":1, // Jumlah thread konkuren.
            "mbps":"12"// Laju pembatasan. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Demo skrip Writer: Menulis file ORC atau Parquet ke OSS

OSS mendukung penulisan file ORC atau Parquet dengan menggunakan kembali HDFS Writer. Selain parameter OSS Writer yang sudah ada, parameter konfigurasi tambahan seperti Path dan FileFormat ditambahkan. Untuk informasi tentang parameter ini, lihat HDFS Writer.

Berikut adalah contoh penulisan file ORC atau Parquet ke OSS:

Berikut hanya contoh. Modifikasi parameter berdasarkan nama dan tipe kolom aktual Anda. Jangan salin dan gunakan contoh ini secara langsung.

• Menulis file ORC ke OSS

  Untuk menulis file ORC, hanya mode skrip yang didukung. Anda harus beralih ke mode skrip untuk konfigurasi. Atur fileFormat ke orc, atur path ke jalur file yang akan ditulis, dan konfigurasi column dalam format {"name":"nama kolom Anda","type": "tipe kolom Anda"}.

  Tipe ORC berikut didukung untuk penulisan:

  Tipe bidang	Tulis batch ke OSS (format ORC)
  TINYINT	Didukung
  SMALLINT	Didukung
  INT	Didukung
  BIGINT	Didukung
  FLOAT	Didukung
  DOUBLE	Didukung
  TIMESTAMP	Didukung
  DATE	Didukung
  VARCHAR	Didukung
  STRING	Didukung
  CHAR	Didukung
  BOOLEAN	Didukung
  DECIMAL	Didukung
  BINARY	Didukung

  {
  "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"
  }
  }

• Menulis ke OSS dalam format 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"
  }
  }

Parameter skrip Writer

message MessageTypeName {
required_or_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;
}