Sync Files from CPFS to OSS Using Streaming Dataflow - Cloud Parallel File Storage - Alibaba Cloud - Cloud Parallel File Storage

You can use streaming tasks to initiate on-demand, file-level data transfers between a CPFS for Lingjun file system and an Object Storage Service (OSS) bucket. Each transfer is triggered by an API call to submit a subtask and does not automatically synchronize file changes.

Overview

Importing and exporting files within a directory requires the following four steps:

Create a dataflow: Map a subdirectory of a CPFS for Lingjun file system to a prefix in an OSS bucket by creating a dataflow.
Create a streaming task: Call the CreateDataFlowTask operation to create a streaming import or export task. This establishes a tunnel between the source and destination directories. After the task is created, it remains in the Running state but does not transfer any data. A streaming subtask is required to start the actual data transfer.
Create a streaming subtask: Call the CreateDataFlowSubTask operation to submit an import or export subtask for each file.
Query the streaming subtask status: Call the DescribeDataFlowSubTask API to query the progress and status of a submitted subtask. When the Status value is COMPLETE and the Progress value is 10000, this indicates that all source data has been successfully transferred to the destination directory.

Prerequisites

A CPFS for Lingjun file system is created. For more information, see Create a file system.
Add a tag with the key cpfs-dataflow and the value true to the destination OSS bucket. Do not delete or modify this tag while the dataflow is active. Otherwise, the CPFS for Lingjun dataflow cannot access the data in the bucket. For more information, see Manage bucket tagging.
Versioning is enabled for the OSS bucket if multiple dataflows export data to it. This prevents data conflicts. For more information, see Overview.

Important

Only CPFS for Lingjun file systems of version 2.6.0 or later support streaming dataflow tasks. For more information about how to check the version of a file system, see the View the version number of a file system section of the "View the details of a file system" topic.

Create a streaming import task

This example shows how to create a streaming import task and a streaming import subtask to import a single file on demand from an OSS bucket to a CPFS for Lingjun file system. In this example, a file from /bmcpfs/test/file.xml in the examplebucket OSS bucket is imported to /oss/mnt/file.xml in the bmcpfs-370jz26fkr2st9**** CPFS for Lingjun file system.

Create a dataflow.

Create a dataflow for the destination file system by calling an API operation or by using the console. After the dataflow is created, note the dataflow ID, such as df-37bae1804cc6****.

Create a dataflow by calling the CreateDataFlow operation.

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

Create a dataflow in the console. For more information, see Manage data flows.

Create a streaming dataflow import task.

Call the CreateDataFlowTask operation to create a streaming dataflow import task, and save the returned TaskId value.

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

Call the CreateDataFlowSubTask operation to submit a 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 the streaming subtask.

Call the DescribeDataFlowSubTasks operation to query the progress and status of a submitted subtask. For the mapping between filter keys and values, see DescribeDataFlowSubTasks.

The following example queries subtask information by dataflow ID.

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

The values of the Progress and Status parameters in the call result indicate the execution progress and status of the subtask. When the task Status is COMPLETE, the task is complete. When the Progress value is 10000, all data has been imported to or exported to the target directory.

Create a streaming export task

This example shows how to create a streaming export task and a streaming export subtask to export a single file on demand from a CPFS for Lingjun file system to an OSS bucket. In this example, a file from /oss_test/yaml/test/file.png in the bmcpfs-370jz26fkr2st9**** CPFS for Lingjun file system is exported to /bmcpfs_test/dataflows/mnt/file.png in the examplebucket OSS bucket.

Create a dataflow.

Create a dataflow for the destination file system by calling an API operation or by using the console. After the dataflow is created, note the dataflow ID, such as df-37bae1804cc6****.

Create a dataflow by calling the CreateDataFlow operation.

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

Create a dataflow in the console. For more information, see Manage data flows.

Create a streaming dataflow export task.

Call the CreateDataFlowTask operation to create a streaming dataflow export task, and save the returned TaskId value.

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

Expected output:

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

Create a streaming export subtask.

Call the CreateDataFlowSubTask operation to submit a 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 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.
}

Expected output:

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

Query the progress and status of the streaming export subtask.

Call the DescribeDataFlowSubTasks operation to query the progress and status of a submitted subtask. For the mapping between filter keys and values, see DescribeDataFlowSubTasks.

The following example queries subtask information by dataflow ID.

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

The values of the Progress and Status parameters in the API response indicate the execution progress and status of the subtask. A Status value of COMPLETE indicates that the task is complete. A Progress value of 10000 indicates that all data has been imported or exported to the target directory.

Related operations

To cancel a streaming subtask, call the CancelDataFlowSubTask operation. Only subtasks in the CREATED or RUNNING state can be canceled.

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