Gestión de tareas de streaming (OpenAPI) - Cloud Parallel File Storage

Las tareas de streaming permiten iniciar transferencias de datos bajo demanda a nivel de archivo entre un sistema de archivos CPFS for Lingjun y un bucket de Object Storage Service (OSS). Cada transferencia se activa mediante una llamada a la API para enviar una subtarea. Los cambios en los archivos no se sincronizan automáticamente.

Descripción general

La importación y exportación de archivos dentro de un directorio requiere los cuatro pasos que se muestran a continuación:

Cree un dataflow: asigne un subdirectorio del sistema de archivos CPFS for Lingjun a un prefijo en un bucket de OSS mediante la creación de un dataflow.
Cree una tarea de streaming: llame a la operación CreateDataFlowTask para crear una tarea de importación o exportación de streaming. Esto establece un túnel entre los directorios de origen y destino. Una vez creada, la tarea permanece en estado Running, pero no transfiere datos. Se requiere una subtarea de streaming para iniciar la transferencia real de datos.
Cree una subtarea de streaming: llame a la operación CreateDataFlowSubTask para enviar una subtarea de importación o exportación por cada archivo.
Consulte el estado de la subtarea de streaming: llame a la API DescribeDataFlowSubTask para consultar el progreso y el estado de una subtarea enviada. Cuando el valor de Status es COMPLETE y el valor de Progress es 10000, significa que todos los datos de origen se transfirieron correctamente al directorio de destino.

Requisitos previos

Debe existir un sistema de archivos CPFS for Lingjun creado. Para obtener más información, consulte Create a file system.
Agregue una etiqueta con la clave cpfs-dataflow y el valor true al bucket de OSS de destino. No elimine ni modifique esta etiqueta mientras el dataflow esté activo. De lo contrario, el dataflow de CPFS for Lingjun no podrá acceder a los datos del bucket. Para obtener más información, consulte Manage bucket tagging.
Si varios dataflows exportan datos al mismo bucket de OSS, habilite el control de versiones en dicho bucket. Esto evita conflictos de datos. Para obtener más información, consulte Overview.

Importante

Solo los sistemas de archivos CPFS for Lingjun de la versión 2.6.0 o posterior admiten tareas de dataflow de streaming. Para obtener más información sobre cómo comprobar la versión de un sistema de archivos, consulte la sección View the version number of a file system del tema "View the details of a file system".

Creación de una tarea de importación de streaming

Este ejemplo muestra cómo crear una tarea y una subtarea de importación de streaming para importar un solo archivo bajo demanda desde un bucket de OSS hacia un sistema de archivos CPFS for Lingjun. En este caso, se importa un archivo desde /bmcpfs/test/file.xml en el bucket de OSS examplebucket hacia /oss/mnt/file.xml en el sistema de archivos CPFS for Lingjun bmcpfs-370jz26fkr2st9****.

Cree un dataflow.

Cree un dataflow para el sistema de archivos de destino mediante una llamada a la API o desde la consola. Una vez creado, anote el identificador del dataflow, por ejemplo, df-37bae1804cc6****.

Cree un dataflow llamando a la operación 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. 
}

Resultado esperado:

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

Cree un dataflow en la consola. Para obtener más información, consulte Manage data flows.

Cree una tarea de importación de dataflow de streaming.

Llame a la operación CreateDataFlowTask para crear una tarea de importación de dataflow de streaming y guarde el valor de TaskId devuelto.

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

Resultado esperado:

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

Cree una subtarea de importación de streaming.

Llame a la operación CreateDataFlowSubTask para enviar una subtarea de importación de 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. 
}

Resultado esperado:

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

Consulte el progreso y el estado de la subtarea de streaming.

Llame a la operación DescribeDataFlowSubTasks para consultar el progreso y el estado de una subtarea enviada. Para conocer la correspondencia entre las claves y los valores de filtro, consulte DescribeDataFlowSubTasks.

En el siguiente ejemplo se consulta la información de la subtarea mediante el identificador del dataflow.

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

Resultado esperado:

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

Los valores de los parámetros Progress y Status en el resultado de la llamada indican el progreso de ejecución y el estado de la subtarea. Cuando el Status de la tarea es COMPLETE, la tarea finalizó. Si el valor de Progress es 10000, todos los datos se importaron o exportaron al directorio de destino.

Creación de una tarea de exportación de streaming

Este ejemplo muestra cómo crear una tarea y una subtarea de exportación de streaming para exportar un solo archivo bajo demanda desde un sistema de archivos CPFS for Lingjun hacia un bucket de OSS. En este caso, se exporta un archivo desde /oss_test/yaml/test/file.png en el sistema de archivos CPFS for Lingjun bmcpfs-370jz26fkr2st9**** hacia /bmcpfs_test/dataflows/mnt/file.png en el bucket de OSS examplebucket.

Cree un dataflow.

Cree un dataflow para el sistema de archivos de destino mediante una llamada a la API o desde la consola. Una vez creado, anote el identificador del dataflow, por ejemplo, df-37bae1804cc6****.

Cree un dataflow llamando a la operación 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. 
}

Resultado esperado:

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

Cree un dataflow en la consola. Para obtener más información, consulte Manage data flows.

Cree una tarea de exportación de dataflow de streaming.

Llame a la operación CreateDataFlowTask para crear una tarea de exportación de dataflow de streaming y guarde el valor de TaskId devuelto.

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

Resultado esperado:

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

Cree una subtarea de exportación de streaming.

Llame a la operación CreateDataFlowSubTask para enviar una subtarea de exportación de 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.
}

Resultado esperado:

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

Consulte el progreso y el estado de la subtarea de exportación de streaming.

En el siguiente ejemplo se consulta la información de la subtarea mediante el identificador del dataflow.

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

Resultado esperado:

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

Los valores de los parámetros Progress y Status en la respuesta de la API indican el progreso de ejecución y el estado de la subtarea. Un valor de Status igual a COMPLETE indica que la tarea finalizó. Un valor de Progress igual a 10000 indica que todos los datos se importaron o exportaron al directorio de destino.

Operaciones relacionadas

Para cancelar una subtarea de streaming, llame a la operación CancelDataFlowSubTask. Solo se pueden cancelar las subtareas que se encuentran en estado CREATED o 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. 
}