ストリーミングタスクを使用すると、CPFS for Lingjun ファイルシステムと Object Storage Service (OSS) バケット間で、オンデマンドのファイルレベルのデータ転送を開始できます。各転送は、サブタスクを送信する API 呼び出しによってトリガーされ、ファイルの変更を自動的に同期することはありません。
概要
ディレクトリ内のファイルのインポートとエクスポートには、次の 4 つのステップが必要です。
-
データフローの作成：データフローを作成して、CPFS for Lingjun ファイルシステムのサブディレクトリを OSS バケットのプレフィックスにマッピングします。
-
ストリーミングタスクの作成：CreateDataFlowTask API を呼び出して、ストリーミングインポートタスクまたはストリーミングエクスポートタスクを作成します。これにより、ソースディレクトリと送信先ディレクトリ間のトンネルが確立されます。タスクを作成すると実行中状態になりますが、この時点ではデータは転送されません。実際のデータ転送を開始するには、ストリーミングサブタスクが必要です。
-
ストリーミングサブタスクの作成：CreateDataFlowSubTask API を呼び出し、ファイルごとにインポートまたはエクスポートのサブタスクを送信します。
ストリーミングサブタスクのステータスを照会する: DescribeDataFlowSubTask API を呼び出して、送信されたサブタスクの進捗とステータスを照会します。ステータスの値が COMPLETE で、進捗 の値が 10000 の場合、すべてのソースデータが宛先ディレクトリに正常に転送されたことを示します。
前提条件
-
CPFS for Lingjun ファイルシステムが作成されます。詳細については、「ファイルシステムの作成」をご参照ください。
宛先の OSS バケットに、キーが
cpfs-dataflowで、値がtrueのタグを追加してください。データフローがアクティブな間は、このタグを削除または変更しないでください。そうしないと、CPFS for Lingjun データフローはバケット内のデータにアクセスできなくなります。詳細については、「バケットのタグ付けの管理」をご参照ください。-
複数のデータフローがデータを OSS バケットにエクスポートすると、そのバケットのバージョニングが有効になります。これにより、データ競合が防止されます。詳細については、「概要」をご参照ください。
ストリーミングデータフロータスクは、バージョン 2.6.0 以降の Lingjun ファイルシステム用の CPFS でのみサポートされます。ファイルシステムのバージョンを確認する方法の詳細については、トピック「ファイルシステムの詳細を表示する」の「ファイルシステムのバージョン番号を表示する」セクションをご参照ください。
ストリーミングインポートタスクの作成
この例では、ストリーミングインポートタスクとストリーミングインポートサブタスクを作成して、OSS バケットから CPFS for Lingjun ファイルシステムに単一のファイルをオンデマンドでインポートする方法を示します。この例では、examplebucket OSS バケットの /bmcpfs/test/file.xml にあるファイルを、bmcpfs-370jz26fkr2st9**** CPFS for Lingjun ファイルシステムの /oss/mnt/file.xml にインポートします。
-
データフローを作成します。
API オペレーションを呼び出すか、コンソールを使用して、送信先ファイルシステムのデータフローを作成します。データフローの作成後、データフロー ID (例：df-37bae1804cc6****) をメモします。
-
CreateDataFlow オペレーションを呼び出して、データフローを作成します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "SourceStorage": "oss://examplebucket", // ソース OSS バケットのエンドポイント。 "FileSystemPath": "/oss/", // OSS にリンクされている CPFS for Lingjun ファイルシステム内のディレクトリ。このディレクトリは既に存在している必要があります。 "SourceStoragePath": "/bmcpfs/", // ソース OSS バケット内のオブジェクトのパス。 }出力例:
{ "RequestId": "473469C7-AA6F-4DC5-B3DB-A3DC0D****3E", "DataFlowId": "df-37bae1804cc6****" } -
コンソールでデータフローを作成します。詳細については、「データフローの管理」をご参照ください。
-
-
ストリーミングデータフローインポートタスクを作成します。
CreateDataFlowTask 操作を呼び出してストリーミングデータフローインポートタスクを作成し、返された
TaskId値を保存します。{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc6****", // データフローの ID。 "TaskAction": "StreamImport", // ストリーミングデータフロータスクのタイプ。インポートは StreamImport、エクスポートは StreamExport を指定します。 "DataType": "MetaAndData", // データタイプ。値を MetaAndData に設定します。 "Directory": "/test/", // 移行するファイルが存在するディレクトリの相対パス。この例では、OSS バケットのプレフィックスを使用します。 "DstDirectory": "/mnt/", // 移行されたファイルが存在するディレクトリの相対パス。この例では、CPFS for Lingjun ファイルシステム内のディレクトリを使用します。 "ConflictPolicy": "SKIP_THE_FILE" // 同じ名前のファイルの競合解決ポリシー。有効な値：OVERWRITE_EXISTING: 同じ名前のファイルを強制的に上書きします。SKIP_THE_FILE: 同じ名前のファイルをスキップします。KEEP_LATEST: 同じ名前のファイルの最新バージョンを保持します。 }出力例:
{ "RequestId": "2D69A58F-345C-4FDE-88E4-BF518948F518", "TaskId": "task-376a61ab2d80****" } -
ストリーミングインポートサブタスクを作成します。
CreateDataFlowSubTask 操作を呼び出して、ストリーミングインポートサブタスクを送信します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc****", // データフローの ID。 "DataFlowTaskId": "task-376a61ab2d80****", // ストリーミングインポートタスクの ID。 "SrcFilePath": "/file.xml", // ストリーミングタスクで指定された Directory 配下のソースファイルの相対パス。 "DstFilePath": "/file.xml" // ストリーミングタスクで指定された DstDirectory 配下の送信先ファイルの相対パス。 }出力例:
{ "RequestId": "A70BEE5D-76D3-49FB-B58F-1F398211A5C3", "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****" } ストリーミングサブタスクの進行状況とステータスを確認します。
DescribeDataFlowSubTasks 操作を呼び出して、送信されたサブタスクの進捗とステータスを照会します。 フィルターキーと値のマッピングについては、「DescribeDataFlowSubTasks」をご参照ください。
次の例では、データフロー ID でサブタスク情報を照会します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "Filters": [ { "Key": "DataFlowIds", "Value": "df-37bae1804cc****" } ] }出力例:
{ "RequestId": "98696EF0-1607-4E9D-B01D-F20930B6****", "DataFlowSubTask": { "DataFlowSubTask": [ { "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc****", // データフローの ID。 "DataFlowTaskId": "task-37b705830bcb****", // ストリーミングデータフロータスクの ID。 "DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****",// ストリーミングデータフローサブタスクの ID。 "SrcFilePath": "/bmcpfs/test/file.xml",// 移行するファイルのパス。 "DstFilePath": "/oss/mnt/file.xml", // 移行されたファイルのパス。 "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": "",// このパラメーターがレスポンスに含まれない、または値が空の場合、エラーは発生していません。 "ProgressStats": { "BytesTotal": 68, "BytesDone": 68, "ActualBytes": 68, "AverageSpeed": 34 }, "FileDetail": { "ModifyTime": 1725897600000000000, "Size": 68, "Checksum": "crc64:850309505450944****"// ファイルのチェックサム。 } } ] } }呼び出し結果の 進捗 および [ステータス] パラメーターの値は、サブタスクの実行進捗とステータスを示します。 タスクの [ステータス] が COMPLETE の場合、タスクは完了です。 進捗 の値が 10000 の場合、すべてのデータがターゲットディレクトリにインポートまたはエクスポートされています。
ストリーミングエクスポートタスクの作成
この例では、ストリーミングエクスポートタスクとストリーミングエクスポートサブタスクを作成して、CPFS for Lingjun ファイルシステムから OSS バケットに単一のファイルをオンデマンドでエクスポートする方法を示します。この例では、bmcpfs-370jz26fkr2st9**** CPFS for Lingjun ファイルシステムの /oss_test/yaml/test/file.png にあるファイルを、examplebucket OSS バケットの /bmcpfs_test/dataflows/mnt/file.png にエクスポートします。
-
データフローを作成します。
API オペレーションを呼び出すか、コンソールを使用して、送信先ファイルシステムのデータフローを作成します。データフローの作成後、データフロー ID (例：df-37bae1804cc6****) をメモします。
-
CreateDataFlow オペレーションを呼び出して、データフローを作成します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "SourceStorage": "oss://examplebucket", // ソース OSS バケットのエンドポイント。 "FileSystemPath": "/oss/", // OSS にリンクされている CPFS for Lingjun ファイルシステム内のディレクトリ。このディレクトリは既に存在している必要があります。 "SourceStoragePath": "/bmcpfs/", // ソース OSS バケット内のオブジェクトのパス。 }出力例:
{ "RequestId": "473469C7-AA6F-4DC5-B3DB-A3DC0D****3E", "DataFlowId": "df-37bae1804cc6****" } -
コンソールでデータフローを作成します。詳細については、「データフローの管理」をご参照ください。
-
-
ストリーミングデータフローエクスポートタスクを作成します。
CreateDataFlowTask 操作を呼び出してストリーミングデータフローエクスポートタスクを作成し、返された
TaskId値を保存します。{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc6****", // データフローの ID。 "TaskAction": "StreamExport", // ストリーミングデータフロータスクのタイプ。エクスポートタスクのため、StreamExport を指定します。 "DataType": "MetaAndData", // データタイプ。値を MetaAndData に設定します。 "Directory": "/yaml/", // 移行するファイルが存在するディレクトリの相対パス。この例では、CPFS for Lingjun ファイルシステム内のディレクトリの相対パスを使用します。 "DstDirectory": "/dataflows/", // 移行されたファイルが存在するディレクトリの相対パス。この例では、OSS バケットのプレフィックスの相対パスを使用します。 "ConflictPolicy": "SKIP_THE_FILE" // 同じ名前のファイルの競合解決ポリシー。有効な値：OVERWRITE_EXISTING: 同じ名前のファイルを強制的に上書きします。SKIP_THE_FILE: 同じ名前のファイルをスキップします。KEEP_LATEST: 同じ名前のファイルの最新バージョンを保持します。 }出力例:
{ "RequestId": "BC7C825C-5F65-4B56-BEF6-98C56C7C930B", "TaskId": "task-37b705830bcb****" } -
ストリーミングエクスポートサブタスクを作成します。
CreateDataFlowSubTask 操作を呼び出して、ストリーミングエクスポートサブタスクを送信します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc****", // データフローの ID。 "DataFlowTaskId": "task-37b705830bcb****", // ストリーミングエクスポートタスクの ID。 "SrcFilePath": "/test/file.png", // ストリーミングタスクで指定された Directory 配下のソースファイルの相対パス。 "DstFilePath": "/mnt/file.png" // ストリーミングタスクで指定された DstDirectory 配下の送信先ファイルの相対パス。 }出力例:
{ "RequestId": "A70BEE5D-76D3-49FB-B58F-1F398211A5C3", "DataFlowSubTaskId": "subTaskId-370l4l3x6qsb1z1****" } ストリーミングエクスポートサブタスクの進行状況とステータスを確認します。
DescribeDataFlowSubTasks 操作を呼び出して、送信されたサブタスクの進捗状況とステータスを照会します。フィルターキーと値のマッピングについては、「DescribeDataFlowSubTasks」をご参照ください。
次の例では、データフロー ID でサブタスク情報を照会します。
{ "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "Filters": [ { "Key": "DataFlowIds", "Value": "df-37bae1804cc****" } ] }出力例:
{ "RequestId": "FCBB356-96CA-135B-84B3-02E6F262B6BD", "DataFlowSubTask": { "DataFlowSubTask": [ { "FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。 "DataFlowId": "df-37bae1804cc****", // データフローの ID。 "DataFlowTaskId": "task-37b705830bcb****", // ストリーミングデータフロータスクの ID。 "DataFlowSubTaskId": "subTaskId-370l4l3x6qsb1z1****",// ストリーミングデータフローサブタスクの ID。 "SrcFilePath": "/oss_test/yaml/test/file.png",// 移行するファイルのパス。 "DstFilePath": "/bmcpfs_test/dataflows/mnt/file.png", // 移行されたファイルのパス。 "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": "",// このパラメーターがレスポンスに含まれない、または値が空の場合、エラーは発生していません。 "ProgressStats": { "BytesTotal": 68, "BytesDone": 68, "ActualBytes": 68, "AverageSpeed": 34 }, "FileDetail": { "ModifyTime": 1725897600000000000, "Size": 68, "Checksum": "crc64:850309505450944****"// ファイルのチェックサム。 } } ] } }API レスポンスの 進捗 および ステータス パラメーターの値は、サブタスクの実行の進捗状況とステータスを示します。ステータスの値が COMPLETE の場合、タスクが完了したことを示します。進捗 の値が 10000 の場合、すべてのデータがターゲットディレクトリにインポートまたはエクスポートされたことを示します。
関連オペレーション
ストリーミングサブタスクをキャンセルするには、CancelDataFlowSubTask 操作を呼び出します。CREATED または RUNNING 状態のサブタスクのみキャンセルできます。
{
"FileSystemId": "bmcpfs-370jz26fkr2st9****", // CPFS for Lingjun ファイルシステムの ID。
"DataFlowId": "df-37bae1804cc****", // データフローの ID。
"DataFlowTaskId": "task-37b705830bcb****", // ストリーミングインポートタスクまたはストリーミングエクスポートタスクの ID。
"DataFlowSubTaskId": "subTaskId-370kyfmyknxcyzw****" // ストリーミングデータフローサブタスクの ID。
}