Praktik terbaik tugas alur data streaming - Cloud Parallel File Storage

Anda dapat membuat streaming task untuk melakukan migrasi file secara berkelanjutan antara sistem file Cloud Parallel File Storage (CPFS) for Lingjun dan bucket Object Storage Service (OSS), satu per satu.

Ikhtisar

Untuk mengimpor dan mengekspor file ke atau dari suatu direktori, lakukan langkah-langkah berikut:

Create a dataflow: Buat dataflow untuk memetakan subdirektori sistem file CPFS for Lingjun ke awalan (prefix) dalam bucket OSS.
Create a streaming task: Panggil operasi CreateDataFlowTask untuk membuat streaming import atau export task. Tugas ini membuat saluran data (tunnel) antara direktori sumber dan tujuan. Setelah streaming task dibuat, statusnya tetap dalam keadaan Running. Namun, tidak ada data yang ditransmisikan melalui saluran tersebut hingga Anda membuat streaming subtask untuk mentransmisikan data.
Create a streaming subtask: Panggil operasi CreateDataFlowSubTask untuk membuat subtask impor atau ekspor untuk setiap file yang berbeda.
Query the status of a streaming subtask: Panggil operasi DescribeDataFlowSubTask untuk menanyakan progres dan status streaming subtask yang telah dikirimkan. Ketika nilai parameter Status adalah COMPLETE dan nilai parameter Progress adalah 10000 dalam hasil respons, seluruh data sumber telah diimpor ke atau diekspor dari direktori tujuan yang ditentukan.

Prasyarat

Sistem file CPFS for Lingjun telah dibuat. Untuk informasi selengkapnya, lihat Create a file system.
Tambahkan tag pada bucket OSS tujuan dengan kunci cpfs-dataflow dan nilai true, seperti yang ditunjukkan pada gambar berikut. Jangan menghapus atau mengubah tag ini selama dataflow sedang digunakan. Jika tidak, dataflow CPFS for Lingjun tidak dapat mengakses data bucket tersebut. Untuk informasi selengkapnya, lihat Manage bucket tagging.
Fitur versioning diaktifkan untuk bucket OSS agar dapat menggunakan beberapa dataflow untuk mengekspor data ke bucket OSS tersebut. Hal ini mencegah konflik data. Untuk informasi selengkapnya, lihat Overview.

Penting

Hanya sistem file CPFS for Lingjun versi V2.6.0 atau yang lebih baru yang mendukung streaming dataflow tasks. Untuk informasi tentang cara melihat versi sistem file, lihat bagian View the version number of a file system dalam topik "View the details of a file system".

Create a streaming import task

Contoh ini menjelaskan cara membuat streaming import job dan streaming import subtask untuk melakukan streaming data secara berkelanjutan dengan granularitas per file. Data dimigrasikan dari subdirektori (`/test/file`) di bawah path /bmcpfs/test/file.xml dalam bucket OSS (`examplebucket`) ke subdirektori (`/mnt/file`) di bawah path /oss/mnt/file.xml dalam sistem file CPFS for Lingjun (`bmcpfs-370jz26fkr2st9****`).

Create a dataflow.

Buat dataflow untuk sistem file tujuan dengan memanggil operasi API atau melalui Konsol. Setelah dataflow dibuat, catat ID-nya, misalnya df-37bae1804cc6****.

Buat dataflow dengan memanggil operasi CreateDataFlow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "SourceStorage": "oss://examplebucket", // The endpoint of the source OSS bucket. 
  "FileSystemPath": "/oss/",  // The directory in the CPFS for Lingjun file system that is linked to OSS. The directory must already exist. 
  "SourceStoragePath": "/bmcpfs/", // The path of the objects in the source OSS bucket. 
}

Expected output:

{
  "RequestId": "473469C7-AA6F-4DC5-B3DB-A3DC0D****3E",
  "DataFlowId": "df-37bae1804cc6****"
}

Buat dataflow di Konsol. Untuk informasi selengkapnya, lihat Manage data flows.

Create a streaming dataflow import task.

Panggil API CreateDataFlowTask untuk membuat streaming dataflow import task dan simpan nilai TaskId yang dikembalikan.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "DataFlowId": "df-37bae1804cc6****", // The ID of the dataflow. 
  "TaskAction": "StreamImport", // The type of the streaming dataflow task. The value is StreamImport for an import task and StreamExport for an export task. 
  "DataType": "MetaAndData", // The data type. Set the value to MetaAndData. 
  "Directory": "/test/", // The relative path of the directory in which the files to be migrated reside. In this example, the prefix of the OSS bucket is used. 
  "DstDirectory": "/mnt/", // The relative path of the directory in which the migrated files reside. In this example, the directory in the CPFS for Lingjun file system is used. 
  "ConflictPolicy": "SKIP_THE_FILE" // The conflict resolution policy for files with the same name. Valid values: OVERWRITE_EXISTING: forcibly overwrites the file with the same name. SKIP_THE_FILE: skips the file with the same name. KEEP_LATEST: keeps the latest version of the file with the same name. 
}

Expected output:

{
  "RequestId": "2D69A58F-345C-4FDE-88E4-BF518948F518",
  "TaskId": "task-376a61ab2d80****"
}

Create a streaming import subtask.

Panggil operasi CreateDataFlowSubTask untuk membuat streaming import subtask.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "DataFlowId": "df-37bae1804cc****", // The ID of the dataflow. 
  "DataFlowTaskId": "task-376a61ab2d80****", // The ID of the streaming import task. 
  "SrcFilePath": "/file.xml", // The path of the file to be migrated in the streaming task. In this example, the path of an object in the OSS bucket is used. 
  "DstFilePath": "/mnt/file.xml" // The path of the migrated file in the streaming task. In this example, the directory in the CPFS for Lingjun file system is used. 
}

Expected output:

{
  "RequestId": "A70BEE5D-76D3-49FB-B58F-1F398211A5C3",
  "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****"
}

Query the progress and status of a streaming subtask.

Panggil operasi DescribeDataFlowSubTasks untuk menanyakan progres dan status streaming subtask yang telah dikirimkan. Kunci yang berbeda sesuai dengan nilai yang berbeda. Untuk informasi selengkapnya, lihat DescribeDataFlowSubTasks.

Contoh ini menjelaskan cara menanyakan informasi subtask berdasarkan ID dataflow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "Filters": [
    {
      "Key": "DataFlowIds",
      "Value": "df-37bae1804cc****"
    }
  ]
}

Expected output:

{
  "RequestId": "98696EF0-1607-4E9D-B01D-F20930B6****",
  "DataFlowSubTask": {
    "DataFlowSubTask": [
      {
        "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
        "DataFlowId": "df-37bae1804cc****", // The ID of the dataflow. 
        "DataFlowTaskId": "task-37b705830bcb****", // The ID of the streaming dataflow task. 
        "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****",// The ID of the streaming dataflow subtask. 
        "SrcFilePath": "/bmcpfs/test/file.xml",// The path of the file to be migrated. 
        "DstFilePath": "/oss/mnt/file.xml", // The path of the migrated file. 
        "Status": "COMPLETE",
        "Progress": 10000,
        "CreateTime": "2024-10-23 16:28:16",
        "StartTime": "2024-10-23 16:28:17",
        "EndTime": "2024-10-23 16:29:22",
        "ErrorMsg": "",// If this parameter is not returned or its returned result is empty, no error occurs. 
        "ProgressStats": {
          "BytesTotal": 68,
          "BytesDone": 68,
          "ActualBytes": 68,
          "AverageSpeed": 34
        },
        "FileDetail": {
          "ModifyTime": 1725897600000000000,
          "Size": 68,
          "Checksum": "crc64:850309505450944****"// The file checksum. 
        }
      }
    ]
  }
}

Nilai parameter Progress dan Status menunjukkan progres dan status streaming subtask. Jika nilai parameter Status adalah COMPLETE, subtask telah selesai. Jika nilai parameter Progress adalah 10000, seluruh data telah diimpor atau diekspor ke direktori tujuan.

Streaming export task

Contoh ini menunjukkan cara membuat streaming export task dan streaming export subtask untuk memigrasikan file file.png dari path /oss_test/yaml/test/ dalam sistem file CPFS for Lingjun bernama `bmcpfs-370jz26fkr2st9****` ke path `/bmcpfs_test/dataflows/mnt/file.png` di bucket OSS bernama `examplebucket`. Proses ini memungkinkan migrasi data berkelanjutan tingkat file.

Create a dataflow.

Buat dataflow untuk sistem file tujuan dengan memanggil operasi API atau melalui Konsol. Setelah dataflow dibuat, catat ID-nya, misalnya df-37bae1804cc6****.

Buat dataflow dengan memanggil operasi CreateDataFlow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "SourceStorage": "oss://examplebucket", // The endpoint of the source OSS bucket. 
  "FileSystemPath": "/oss/",  // The directory in the CPFS for Lingjun file system that is linked to OSS. The directory must already exist. 
  "SourceStoragePath": "/bmcpfs/", // The path of the objects in the source OSS bucket. 
}

Expected output:

{
  "RequestId": "473469C7-AA6F-4DC5-B3DB-A3DC0D****3E",
  "DataFlowId": "df-37bae1804cc6****"
}

Buat dataflow di Konsol. Untuk informasi selengkapnya, lihat Manage data flows.

Create a streaming dataflow export task.

Panggil API CreateDataFlowTask untuk membuat streaming dataflow export task dan simpan nilai TaskId yang dikembalikan.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "DataFlowId": "df-37bae1804cc6****", // The ID of the dataflow. 
  "TaskAction": "StreamImport", // The type of the streaming dataflow task. The value is StreamExport for an export task. 
  "DataType": "MetaAndData", // The data type. Set the value to MetaAndData. 
  "Directory": "/yaml/", // The relative path of the directory in which the files to be migrated reside. In this example, the relative path of the directory in the CPFS for Lingjun file system is used. 
  "DstDirectory": "/dataflows/", // The relative path of the directory in which the migrated files reside. In this example, the relative path of the OSS bucket prefix is used. 
  "ConflictPolicy": "SKIP_THE_FILE" // The conflict resolution policy for files with the same name. Valid values: OVERWRITE_EXISTING: forcibly overwrites the file with the same name. SKIP_THE_FILE: skips the file with the same name. KEEP_LATEST: keeps the latest version of the file with the same name. 
}

Expected output:

{
  "RequestId": "BC7C825C-5F65-4B56-BEF6-98C56C7C930B",
  "TaskId": "task-37b705830bcb****"
}

Create a streaming export subtask.

Panggil operasi CreateDataFlowSubTask untuk membuat streaming export subtask.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "DataFlowId": "df-37bae1804cc****", // The ID of the dataflow. 
  "DataFlowTaskId": "task-37b705830bcb****", //The ID of the streaming export task. 
  "SrcFilePath": "/test/file.png", // The relative path of the directory in which the file to be migrated resides in the streaming task. 
  "DstFilePath": "/mnt/file.png" // The relative path of the directory in which the migrated file resides in the streaming task. 
}

Expected output:

{
  "RequestId": "A70BEE5D-76D3-49FB-B58F-1F398211A5C3",
  "DataFlowSubTaskId": "subTaskId-370l4l3x6qsb1z1****"
}

Query the progress and status of a streaming export subtask.

Contoh ini menjelaskan cara menanyakan informasi subtask berdasarkan ID dataflow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "Filters": [
    {
      "Key": "DataFlowIds",
      "Value": "df-37bae1804cc****"
    }
  ]
}

Expected output:

{
  "RequestId": "FCBB356-96CA-135B-84B3-02E6F262B6BD",
  "DataFlowSubTask": {
    "DataFlowSubTask": [
      {
        "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
        "DataFlowId": "df-37bae1804cc****", // The ID of the dataflow. 
        "DataFlowTaskId": "task-37b705830bcb****", // The ID of the streaming dataflow task. 
        "DataFlowSubTaskId": "subTaskId-370l4l3x6qsb1z1****",// The ID of the streaming dataflow subtask. 
        "SrcFilePath": "/oss_test/yaml/test/file.png",// The path of the file to be migrated. 
        "DstFilePath": "/bmcpfs_test/dataflows/mnt/file.png", // The path of the migrated file. 
        "Status": "COMPLETE",
        "Progress": 10000,
        "CreateTime": "2024-10-23 17:18:16",
        "StartTime": "2024-10-23 17:18:17",
        "EndTime": "2024-10-23 17:19:00",
        "ErrorMsg": "",// If this parameter is not returned or its returned result is empty, no error occurs. 
        "ProgressStats": {
          "BytesTotal": 68,
          "BytesDone": 68,
          "ActualBytes": 68,
          "AverageSpeed": 34
        },
        "FileDetail": {
          "ModifyTime": 1725897600000000000,
          "Size": 68,
          "Checksum": "crc64:850309505450944****"// The file checksum. 
        }
      }
    ]
  }
}

Operasi terkait

Untuk membatalkan streaming subtask, panggil operasi CancelDataFlowSubTask. Anda hanya dapat membatalkan streaming subtask yang berada dalam status CREATED atau RUNNING.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // The ID of the CPFS for Lingjun file system. 
  "DataFlowId": "df-37bae1804cc****", // The ID of the dataflow. 
  "DataFlowTaskId": "task-37b705830bcb****", // The ID of the streaming import or export task. 
  "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****" // The ID of the streaming dataflow subtask. 
}