Sinkronisasi File dari CPFS ke OSS Menggunakan Streaming Dataflow - Cloud Parallel File Storage - Alibaba Cloud - Cloud Parallel File Storage

Anda dapat menggunakan tugas streaming untuk memulai transfer data tingkat file sesuai permintaan antara sistem file CPFS for Lingjun dan bucket Object Storage Service (OSS). Setiap transfer dipicu oleh panggilan API untuk mengirimkan subtask dan tidak secara otomatis menyinkronkan perubahan file.

Ikhtisar

Mengimpor dan mengekspor file dalam suatu direktori mencakup empat langkah berikut:

Buat dataflow: Petakan subdirektori sistem file CPFS for Lingjun ke awalan (prefix) dalam bucket OSS dengan membuat dataflow.
Buat tugas streaming: Panggil operasi CreateDataFlowTask untuk membuat tugas impor atau ekspor streaming. Ini akan membuat saluran data antara direktori sumber dan tujuan. Setelah tugas dibuat, statusnya tetap Running namun tidak mentransfer data apa pun. Diperlukan subtask streaming untuk memulai transfer data yang sebenarnya.
Buat subtask streaming: Panggil operasi CreateDataFlowSubTask untuk mengirimkan subtask impor atau ekspor untuk setiap file.
Kueri status subtask streaming: Panggil API DescribeDataFlowSubTask untuk mengecek progres dan status subtask yang telah dikirimkan. Ketika nilai Status adalah COMPLETE dan nilai Progress adalah 10000, hal ini menunjukkan bahwa seluruh data sumber telah berhasil ditransfer ke direktori tujuan.

Prasyarat

Sistem file CPFS for Lingjun telah dibuat. Untuk informasi lebih lanjut, lihat Buat sistem file.
Tambahkan tag dengan kunci cpfs-dataflow dan nilai true ke bucket OSS tujuan. Jangan menghapus atau mengubah tag ini selama dataflow aktif. Jika tidak, dataflow CPFS for Lingjun tidak dapat mengakses data dalam bucket tersebut. Untuk informasi lebih lanjut, lihat Kelola penandaan bucket.
Versi bucket OSS harus diaktifkan jika beberapa dataflow mengekspor data ke bucket tersebut. Hal ini mencegah konflik data. Untuk informasi lebih lanjut, lihat Ikhtisar.

Penting

Hanya sistem file CPFS for Lingjun versi 2.6.0 atau yang lebih baru yang mendukung tugas dataflow streaming. Untuk informasi lebih lanjut tentang cara memeriksa versi sistem file, lihat bagian Lihat nomor versi sistem file pada topik "Lihat detail sistem file".

Buat tugas impor streaming

Contoh ini menunjukkan cara membuat tugas impor streaming dan subtask impor streaming untuk mengimpor satu file sesuai permintaan dari bucket OSS ke sistem file CPFS for Lingjun. Pada contoh ini, file dari /bmcpfs/test/file.xml dalam bucket OSS examplebucket diimpor ke /oss/mnt/file.xml dalam sistem file CPFS for Lingjun bmcpfs-370jz26fkr2st9****.

Buat dataflow.

Buat dataflow untuk sistem file tujuan dengan memanggil operasi API atau menggunakan Konsol. Setelah dataflow dibuat, catat ID dataflow-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. 
}

Output yang diharapkan:

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

Buat dataflow di Konsol. Untuk informasi lebih lanjut, lihat Kelola data flow.

Buat tugas impor dataflow streaming.

Panggil operasi CreateDataFlowTask untuk membuat tugas impor dataflow streaming, 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. 
}

Output yang diharapkan:

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

Buat subtask impor streaming.

Panggil operasi CreateDataFlowSubTask untuk mengirimkan subtask impor streaming.

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

Output yang diharapkan:

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

Kueri progres dan status subtask streaming.

Panggil operasi DescribeDataFlowSubTasks untuk mengecek progres dan status subtask yang telah dikirimkan. Untuk pemetaan antara kunci filter dan nilainya, lihat DescribeDataFlowSubTasks.

Contoh berikut mengecek informasi subtask berdasarkan ID dataflow.

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

Output yang diharapkan:

{
  "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 dalam hasil panggilan menunjukkan progres eksekusi dan status subtask. Ketika Status tugas adalah COMPLETE, tugas telah selesai. Ketika nilai Progress adalah 10000, seluruh data telah diimpor atau diekspor ke direktori target.

Buat tugas ekspor streaming

Contoh ini menunjukkan cara membuat tugas ekspor streaming dan subtask ekspor streaming untuk mengekspor satu file sesuai permintaan dari sistem file CPFS for Lingjun ke bucket OSS. Pada contoh ini, file dari /oss_test/yaml/test/file.png dalam sistem file CPFS for Lingjun bmcpfs-370jz26fkr2st9**** diekspor ke /bmcpfs_test/dataflows/mnt/file.png dalam bucket OSS examplebucket.

Buat dataflow.

Buat dataflow untuk sistem file tujuan dengan memanggil operasi API atau menggunakan Konsol. Setelah dataflow dibuat, catat ID dataflow-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. 
}

Output yang diharapkan:

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

Buat dataflow di Konsol. Untuk informasi lebih lanjut, lihat Kelola data flow.

Buat tugas ekspor dataflow streaming.

Panggil operasi CreateDataFlowTask untuk membuat tugas ekspor dataflow streaming, 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": "StreamExport", // 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. 
}

Output yang diharapkan:

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

Buat subtask ekspor streaming.

Panggil operasi CreateDataFlowSubTask untuk mengirimkan subtask ekspor streaming.

{
  "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 source file under the Directory specified in the streaming task.
  "DstFilePath": "/mnt/file.png" // The relative path of the destination file under the DstDirectory specified in the streaming task.
}

Output yang diharapkan:

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

Kueri progres dan status subtask ekspor streaming.

Panggil operasi DescribeDataFlowSubTasks untuk mengecek progres dan status subtask yang telah dikirimkan. Untuk pemetaan antara kunci filter dan nilainya, lihat DescribeDataFlowSubTasks.

Contoh berikut mengecek informasi subtask berdasarkan ID dataflow.

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

Output yang diharapkan:

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

Nilai parameter Progress dan Status dalam respons API menunjukkan progres eksekusi dan status subtask. Nilai Status COMPLETE menunjukkan bahwa tugas telah selesai. Nilai Progress sebesar 10000 menunjukkan bahwa seluruh data telah diimpor atau diekspor ke direktori target.

Operasi terkait

Untuk membatalkan subtask streaming, panggil operasi CancelDataFlowSubTask. Hanya subtask dalam status CREATED atau RUNNING yang dapat dibatalkan.

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