Praktik terbaik tugas aliran data streaming - Cloud Parallel File Storage

Anda dapat membuat tugas streaming untuk memigrasikan file secara terus-menerus antara Cloud Parallel File Storage (CPFS) untuk sistem file Lingjun dan Bucket OSS.

Ikhtisar

Untuk mengimpor dan mengekspor file ke dan dari direktori, ikuti langkah-langkah berikut:

Buat Aliran Data: Buat aliran data untuk memetakan subdirektori CPFS untuk sistem file Lingjun ke awalan dalam Bucket OSS.
Buat Tugas Streaming: Panggil operasi CreateDataFlowTask untuk membuat tugas impor atau ekspor streaming. Tugas ini membentuk terowongan antara direktori sumber dan tujuan. Setelah dibuat, tugas tetap dalam status Running, namun tidak ada data yang ditransmisikan melalui terowongan hingga subtugas streaming dibuat.
Buat Subtugas Streaming: Panggil operasi CreateDataFlowSubTask untuk membuat subtugas impor atau ekspor untuk setiap file.
Kueri Status Subtugas Streaming: Panggil operasi DescribeDataFlowSubTask untuk menanyakan kemajuan dan status subtugas streaming. Ketika nilai parameter Status adalah COMPLETE dan nilai parameter Progress adalah 10000, semua data telah diimpor atau diekspor ke direktori tujuan.

Prasyarat

Sistem file CPFS untuk Lingjun telah dibuat. Untuk informasi lebih lanjut, lihat Buat Sistem File.
Tag dengan kunci cpfs-dataflow dan nilai true telah ditambahkan ke Bucket OSS tujuan. Jangan hapus atau ubah tag tersebut saat aliran data digunakan. Jika tidak, aliran data CPFS untuk Lingjun tidak dapat mengakses data di Bucket OSS. Untuk informasi lebih lanjut, lihat Kelola Tag Bucket.
Aktifkan fitur pengelolaan versi untuk Bucket OSS jika Anda perlu menggunakan beberapa aliran data untuk mengekspor data ke Bucket OSS. Ini mencegah konflik data. Untuk informasi lebih lanjut, lihat Ikhtisar.

Penting

Hanya sistem file CPFS untuk Lingjun versi V2.6.0 atau lebih baru yang mendukung tugas aliran data streaming. Untuk informasi lebih lanjut tentang cara melihat versi sistem file, lihat bagian Lihat Nomor Versi Sistem File dari topik "Lihat Detail Sistem File".

Buat tugas impor streaming

Contoh ini menjelaskan cara membuat tugas impor streaming dan subtugas impor streaming untuk memigrasikan file file.xml di direktori /bmcpfs/test/ dari bucket OSS bernama examplebucket ke direktori /oss/mnt/ pada sistem file CPFS untuk Lingjun bernama bmcpfs-370jz26fkr2st9****. Anda dapat menggunakan contoh ini sebagai referensi untuk memigrasikan file secara terus-menerus satu per satu.

Buat Aliran Data.

Buat aliran data untuk sistem file tujuan dengan memanggil operasi API atau melakukan operasi di konsol. Setelah aliran data dibuat, peroleh ID aliran data, seperti df-37bae1804cc6****.

Buat aliran data dengan memanggil operasi CreateDataFlow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "SourceStorage": "oss://examplebucket", // Titik akhir bucket OSS sumber.
  "FileSystemPath": "/oss/",  // Direktori dalam sistem file CPFS untuk Lingjun yang terhubung ke OSS. Direktori harus sudah ada.
  "SourceStoragePath": "/bmcpfs/", // Jalur objek dalam bucket OSS sumber.
}

Output yang Diharapkan:

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

Buat aliran data di konsol. Untuk informasi lebih lanjut, lihat Kelola Aliran Data.

Buat Tugas Impor Aliran Data Streaming.

Panggil operasi CreateDataFlowTask untuk membuat tugas impor aliran data streaming, dan simpan nilai yang dikembalikan untuk parameter TaskId.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "DataFlowId": "df-37bae1804cc6****", // ID aliran data.
  "TaskAction": "StreamImport", // Jenis tugas aliran data streaming. Nilai StreamImport untuk tugas impor dan StreamExport untuk tugas ekspor.
  "DataType": "MetaAndData", // Jenis data. Atur nilainya menjadi MetaAndData.
  "Directory": "/test/", // Jalur relatif direktori tempat file yang akan dimigrasikan berada. Dalam contoh ini, awalan bucket OSS digunakan.
  "DstDirectory": "/mnt/", // Jalur relatif direktori tempat file yang dimigrasikan berada. Dalam contoh ini, direktori dalam sistem file CPFS untuk Lingjun digunakan.
  "ConflictPolicy": "SKIP_THE_FILE" // Kebijakan resolusi konflik untuk file dengan nama yang sama. Nilai valid: OVERWRITE_EXISTING: paksa menimpa file dengan nama yang sama. SKIP_THE_FILE: lewati file dengan nama yang sama. KEEP_LATEST: pertahankan versi terbaru dari file dengan nama yang sama.
}

Output yang Diharapkan:

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

Buat Subtugas Impor Streaming.

Panggil operasi CreateDataFlowSubTask untuk membuat subtugas impor streaming.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "DataFlowId": "df-37bae1804cc****", // ID aliran data.
  "DataFlowTaskId": "task-376a61ab2d80****", // ID tugas impor streaming.
  "SrcFilePath": "/file.xml", // Jalur file yang akan dimigrasikan dalam tugas streaming. Dalam contoh ini, jalur objek dalam bucket OSS digunakan.
  "DstFilePath": "/mnt/file.xml" // Jalur file yang dimigrasikan dalam tugas streaming. Dalam contoh ini, direktori dalam sistem file CPFS untuk Lingjun digunakan.
}

Output yang Diharapkan:

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

Kueri Kemajuan dan Status Subtugas Streaming.

Panggil operasi DescribeDataFlowSubTasks untuk menanyakan kemajuan dan status subtugas streaming yang telah dikirimkan. Kunci yang berbeda sesuai dengan nilai yang berbeda. Untuk informasi lebih lanjut, lihat DescribeDataFlowSubTasks.

Contoh ini menjelaskan cara menanyakan informasi tentang subtugas berdasarkan ID aliran data.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "Filters": [
    {
      "Key": "DataFlowIds",
      "Value": "df-37bae1804cc****"
    }
  ]
}

Output yang Diharapkan:

{
  "RequestId": "98696EF0-1607-4E9D-B01D-F20930B6****",
  "DataFlowSubTask": {
    "DataFlowSubTask": [
      {
        "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
        "DataFlowId": "df-37bae1804cc****", // ID aliran data.
        "DataFlowTaskId": "task-37b705830bcb****", // ID tugas aliran data streaming.
        "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****",// ID subtugas aliran data streaming.
        "SrcFilePath": "/bmcpfs/test/file.xml",// Jalur file yang akan dimigrasikan.
        "DstFilePath": "/oss/mnt/file.xml", // Jalur file yang dimigrasikan.
        "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": "",// Jika parameter ini tidak dikembalikan atau hasilnya kosong, tidak ada kesalahan yang terjadi.
        "ProgressStats": {
          "BytesTotal": 68,
          "BytesDone": 68,
          "ActualBytes": 68,
          "AverageSpeed": 34
        },
        "FileDetail": {
          "ModifyTime": 1725897600000000000,
          "Size": 68,
          "Checksum": "crc64:850309505450944****"// Checksum file.
        }
      }
    ]
  }
}

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

Buat tugas ekspor streaming

Contoh ini menjelaskan cara membuat tugas ekspor streaming dan subtugas ekspor streaming untuk memigrasikan file file.png dari direktori /oss_test/yaml/test/ dalam sistem file CPFS untuk Lingjun bernama bmcpfs-370jz26fkr2st9**** ke direktori /bmcpfs_test/dataflows/mnt/ di bucket OSS bernama examplebucket. Anda dapat menggunakan contoh ini sebagai referensi untuk memigrasikan file secara terus-menerus satu per satu.

Buat Aliran Data.

Buat aliran data untuk sistem file tujuan dengan memanggil operasi API atau melakukan operasi di konsol. Setelah aliran data dibuat, peroleh ID aliran data, seperti df-37bae1804cc6****.

Buat aliran data dengan memanggil operasi CreateDataFlow.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "SourceStorage": "oss://examplebucket", // Titik akhir bucket OSS sumber.
  "FileSystemPath": "/oss/",  // Direktori dalam sistem file CPFS untuk Lingjun yang terhubung ke OSS. Direktori harus sudah ada.
  "SourceStoragePath": "/bmcpfs/", // Jalur objek dalam bucket OSS sumber.
}

Output yang Diharapkan:

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

Buat aliran data di konsol. Untuk informasi lebih lanjut, lihat Kelola Aliran Data.

Buat Tugas Ekspor Aliran Data Streaming.

Panggil operasi CreateDataFlowTask untuk membuat tugas ekspor aliran data streaming, dan simpan nilai yang dikembalikan untuk parameter TaskId.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "DataFlowId": "df-37bae1804cc6****", // ID aliran data.
  "TaskAction": "StreamImport", // Jenis tugas aliran data streaming. Nilai StreamExport untuk tugas ekspor.
  "DataType": "MetaAndData", // Jenis data. Atur nilainya menjadi MetaAndData.
  "Directory": "/yaml/", // Jalur relatif direktori tempat file yang akan dimigrasikan berada. Dalam contoh ini, jalur relatif direktori dalam sistem file CPFS untuk Lingjun digunakan.
  "DstDirectory": "/dataflows/", // Jalur relatif direktori tempat file yang dimigrasikan berada. Dalam contoh ini, jalur relatif awalan bucket OSS digunakan.
  "ConflictPolicy": "SKIP_THE_FILE" // Kebijakan resolusi konflik untuk file dengan nama yang sama. Nilai valid: OVERWRITE_EXISTING: paksa menimpa file dengan nama yang sama. SKIP_THE_FILE: lewati file dengan nama yang sama. KEEP_LATEST: pertahankan versi terbaru dari file dengan nama yang sama.
}

Output yang Diharapkan:

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

Buat Subtugas Ekspor Streaming.

Panggil operasi CreateDataFlowSubTask untuk membuat subtugas ekspor streaming.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "DataFlowId": "df-37bae1804cc****", // ID aliran data.
  "DataFlowTaskId": "task-37b705830bcb****", //ID tugas ekspor streaming.
  "SrcFilePath": "/test/file.png", // Jalur relatif direktori tempat file yang akan dimigrasikan berada dalam tugas streaming.
  "DstFilePath": "/mnt/file.png" // Jalur relatif direktori tempat file yang dimigrasikan berada dalam tugas streaming.
}

Output yang Diharapkan:

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

Kueri Kemajuan dan Status Subtugas Ekspor Streaming.

Contoh ini menjelaskan cara menanyakan informasi tentang subtugas berdasarkan ID aliran data.

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "Filters": [
    {
      "Key": "DataFlowIds",
      "Value": "df-37bae1804cc****"
    }
  ]
}

Output yang Diharapkan:

{
  "RequestId": "FCBB356-96CA-135B-84B3-02E6F262B6BD",
  "DataFlowSubTask": {
    "DataFlowSubTask": [
      {
        "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
        "DataFlowId": "df-37bae1804cc****", // ID aliran data.
        "DataFlowTaskId": "task-37b705830bcb****", // ID tugas aliran data streaming.
        "DataFlowSubTaskId": "subTaskId-370l4l3x6qsb1z1****",// ID subtugas aliran data streaming.
        "SrcFilePath": "/oss_test/yaml/test/file.png",// Jalur file yang akan dimigrasikan.
        "DstFilePath": "/bmcpfs_test/dataflows/mnt/file.png", // Jalur file yang dimigrasikan.
        "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": "",// Jika parameter ini tidak dikembalikan atau hasilnya kosong, tidak ada kesalahan yang terjadi.
        "ProgressStats": {
          "BytesTotal": 68,
          "BytesDone": 68,
          "ActualBytes": 68,
          "AverageSpeed": 34
        },
        "FileDetail": {
          "ModifyTime": 1725897600000000000,
          "Size": 68,
          "Checksum": "crc64:850309505450944****"// Checksum file.
        }
      }
    ]
  }
}

Apa yang harus dilakukan selanjutnya

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

{
  "FileSystemId": "bmcpfs-370jz26fkr2st9****", // ID sistem file CPFS untuk Lingjun.
  "DataFlowId": "df-37bae1804cc****", // ID aliran data.
  "DataFlowTaskId": "task-37b705830bcb****", // ID tugas impor atau ekspor streaming.
  "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****" // ID subtugas aliran data streaming.
}