Gerencie tarefas de streaming (API) - Cloud Parallel File Storage: Central de Documentação do Alibaba Cloud

Use tarefas de streaming para iniciar transferências de dados sob demanda, no nível de arquivo, entre um sistema de arquivos CPFS for Lingjun e um bucket do Object Storage Service (OSS). Cada transferência é acionada por uma chamada de API para enviar uma subtarefa e não sincroniza automaticamente as alterações nos arquivos.

Visão geral

A importação e a exportação de arquivos em um diretório exigem as quatro etapas a seguir:

Crie um dataflow: mapeie um subdiretório de um sistema de arquivos CPFS for Lingjun para um prefixo em um bucket do OSS ao criar um dataflow.
Crie uma tarefa de streaming: chame a operação CreateDataFlowTask para criar uma tarefa de importação ou exportação de streaming. Essa ação estabelece um túnel entre os diretórios de source e destino. Após a criação, a tarefa permanece no estado Running, mas não transfere dados. É necessária uma subtarefa de streaming para iniciar a transferência real dos dados.
Crie uma subtarefa de streaming: chame a operação CreateDataFlowSubTask para enviar uma subtarefa de importação ou exportação para cada arquivo.
Visualize o status da subtarefa de streaming: chame a API DescribeDataFlowSubTask para visualizar o progresso e o status de uma subtarefa enviada. Se o valor de Status for COMPLETE e o valor de Progress for 10000, todos os dados de source foram transferidos com sucesso para o diretório de destino.

Pré-requisitos

Você deve ter criado um sistema de arquivos CPFS for Lingjun. Para mais informações, consulte Criar um sistema de arquivos.
Adicione uma tag com a chave cpfs-dataflow e o valor true ao bucket do OSS de destino. Não exclua nem modifique essa tag enquanto o dataflow estiver ativo. Caso contrário, o dataflow do CPFS for Lingjun não conseguirá acessar os dados no bucket. Para mais informações, consulte Gerenciar tags de bucket.
Ative o versionamento no bucket do OSS se vários dataflows exportarem dados para ele. Isso evita conflitos de dados. Para mais informações, consulte Visão geral.

Importante

Somente sistemas de arquivos CPFS for Lingjun da versão 2.6.0 ou posterior suportam tarefas de dataflow de streaming. Para mais informações sobre como verificar a versão de um sistema de arquivos, consulte a seção Visualizar o número da versão de um sistema de arquivos no tópico "Visualizar os detalhes de um sistema de arquivos".

Criar uma tarefa de importação de streaming

Este exemplo demonstra como criar uma tarefa e uma subtarefa de importação de streaming para importar um único arquivo sob demanda de um bucket do OSS para um sistema de arquivos CPFS for Lingjun. Neste cenário, um arquivo localizado em /bmcpfs/test/file.xml no bucket do OSS examplebucket é importado para /oss/mnt/file.xml no sistema de arquivos CPFS for Lingjun bmcpfs-370jz26fkr2st9****.

Crie um dataflow.

Crie um dataflow para o sistema de arquivos de destino chamando uma operação de API ou usando o console. Após criar o dataflow, anote o ID do dataflow, como df-37bae1804cc6****.

Chame a operação CreateDataFlow para criar um dataflow.

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

Saída esperada:

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

Crie um dataflow no console. Para mais informações, consulte Gerenciar fluxos de dados.

Crie uma tarefa de importação de dataflow de streaming.

Chame a operação CreateDataFlowTask para criar uma tarefa de importação de dataflow de streaming e salve o valor TaskId retornado.

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

Saída esperada:

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

Crie uma subtarefa de importação de streaming.

Chame a operação CreateDataFlowSubTask para enviar uma subtarefa de importação 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. 
}

Saída esperada:

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

Visualize o progresso e o status da subtarefa de streaming.

Chame a operação DescribeDataFlowSubTasks para visualizar o progresso e o status de uma subtarefa enviada. Para obter o mapeamento entre chaves e valores de filtro, consulte DescribeDataFlowSubTasks.

O exemplo a seguir consulta informações da subtarefa pelo ID do dataflow.

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

Saída esperada:

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

Os valores dos parâmetros Progress e Status no resultado da chamada indicam o progresso e o status de execução da subtarefa. Quando o Status da tarefa for COMPLETE, a tarefa estará concluída. Quando o valor de Progress for 10000, todos os dados terão sido importados ou exportados para o diretório de destino.

Criar uma tarefa de exportação de streaming

Este exemplo demonstra como criar uma tarefa e uma subtarefa de exportação de streaming para exportar um único arquivo sob demanda de um sistema de arquivos CPFS for Lingjun para um bucket do OSS. Neste cenário, um arquivo localizado em /oss_test/yaml/test/file.png no sistema de arquivos CPFS for Lingjun bmcpfs-370jz26fkr2st9**** é exportado para /bmcpfs_test/dataflows/mnt/file.png no bucket do OSS examplebucket.

Crie um dataflow.

Crie um dataflow para o sistema de arquivos de destino chamando uma operação de API ou usando o console. Após criar o dataflow, anote o ID do dataflow, como df-37bae1804cc6****.

Chame a operação CreateDataFlow para criar um dataflow.

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

Saída esperada:

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

Crie um dataflow no console. Para mais informações, consulte Gerenciar fluxos de dados.

Crie uma tarefa de exportação de dataflow de streaming.

Chame a operação CreateDataFlowTask para criar uma tarefa de exportação de dataflow de streaming e salve o valor TaskId retornado.

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

Saída esperada:

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

Crie uma subtarefa de exportação de streaming.

Chame a operação CreateDataFlowSubTask para enviar uma subtarefa de exportação 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.
}

Saída esperada:

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

Visualize o progresso e o status da subtarefa de exportação de streaming.

O exemplo a seguir consulta informações da subtarefa pelo ID do dataflow.

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

Saída esperada:

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

Os valores dos parâmetros Progress e Status na resposta da API indicam o progresso e o status de execução da subtarefa. Um valor de Status igual a COMPLETE indica que a tarefa foi concluída. Um valor de Progress igual a 10000 indica que todos os dados foram importados ou exportados para o diretório de destino.

Operações relacionadas

Para cancelar uma subtarefa de streaming, chame a operação CancelDataFlowSubTask. Somente subtarefas no estado CREATED ou RUNNING podem ser canceladas.

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