OpenLake:Migrar DolphinScheduler para DataWorks

# DolphinScheduler 1.x
curl -H "token:<YourToken>" -X GET http://<YourIp>:12345/dolphinscheduler/projects/query-project-list

# DolphinScheduler 2.x
curl -H "token:<YourToken>" -X GET http://<YourIp>:12345/dolphinscheduler/projects/list

# DolphinScheduler 3.x
curl -H "token:<YourToken>" -X GET http://<YourIp>:12345/dolphinscheduler/projects/list

2. Configurar informações de conexão

Na pasta conf, crie um arquivo de configuração JSON (por exemplo, read.json).

• Remova os comentários do arquivo JSON antes de usá-lo.

{
  "schedule_datasource": {
    "name": "YourDolphin", // Give your DolphinScheduler data source a name.
    "type": "DolphinScheduler", // Data source type (DolphinScheduler)
    "properties": {
      "endpoint": "http://localhost:12345", // Endpoint
      "project": "Comprehensive Test", // Project name
      "token": "***********************" // Token
    },
    "operaterType": "AUTO" // Connection type (AUTO: Automatically get scheduling information through the API)
  },
  "conf": {

  }
}

sh ./bin/run.sh read \
-c ./conf/<your_config_file>.JSON \
-f ./data/0_OriginalPackage/<api_raw_info_package>.zip \
-o ./data/1_ReaderOutput/<source_discovery_export_package>.zip \
-t <PluginName>

sh ./bin/run.sh read \
-c ./conf/projectA_read.JSON \
-f ./data/0_OriginalPackage/projectA_ApiOutput.zip \
-o ./data/1_ReaderOutput/projectA_ReaderOutput.zip \
-t dolphinv3-reader

5,1. (Descoberta em lote) Posso descobrir vários projetos de uma vez?

Sim. Insira vários nomes de projetos no item de configuração project, separados por vírgulas. Não adicione espaços entre os nomes, pois os nomes de projetos do DolphinScheduler podem conter espaços e a ferramenta trata os espaços como parte do nome.

• Antes de usar o arquivo, remova os comentários do código JSON.

{
  "schedule_datasource": {
    "name": "YourDolphin", // Name your Dolphin data source.
    "type": "DolphinScheduler", // Data source type (DolphinScheduler)
    "properties": {
      "endpoint": "http://localhost:12345", // Endpoint
      "project": "Project1,Project2", // Project name
      "token": "***********************" // Token
    },
    "operaterType": "AUTO" // Connection type (AUTO: Automatically obtain scheduling information through the API)
  },
  "conf": {

  }
}

No comando de execução, os parâmetros -f e -o devem especificar um caminho de pasta. A ferramenta cria automaticamente um pacote de exportação separado para cada projeto.

sh ./bin/run.sh read \
-c ./conf/<your_config_file>.JSON \
-f ./data/0_OriginalPackage/ \
-o ./data/1_ReaderOutput/ \
-t <dolphinv1/2/3-reader>

5,2. (Modo manual) E se não houver API?

Caso o módulo de API tenha sido removido do DolphinScheduler, construa manualmente o pacote de informações brutas no diretório ./data/0_OriginalPackage/ e defina operaterType como MANUAL na configuração. A ferramenta usará então o pacote construído manualmente como entrada.

• Antes de usar o arquivo, remova os comentários do código JSON.

{
  "schedule_datasource": {
    "name": "YourDolphin", // Name your Dolphin data source.
    "type": "DolphinScheduler", // Data source type (DolphinScheduler)
    "properties": {
      "endpoint": "http://localhost:12345", // Endpoint
      "project": "ComprehensiveTest", // Project name
      "token": "***********************" // Token
    },
    "operaterType": "MANUAL" // Connection type (MANUAL: Offline mode)
  },
  "conf": {

  }
}

Exemplo da estrutura do pacote bruto:

.
├── package_info.JSON
├── projects.JSON
├── projects
│   └── ComprehensiveTest
│       └── processDefinition
│           └── process_definitions_page_1.JSON
├── datasource
│   └── datasource_page_1.JSON
├── resource
│   └── resources.JSON
└── udfFunction
    └── udf_function_page_1.JSON

O arquivo package_info.json contém informações do pacote, incluindo a versão do DolphinScheduler.

{
  "version": "3.2.0"
}

O arquivo projects.json contém informações do projeto. Ao construí-lo manualmente, concentre-se em preencher os campos id, userId, code e name.

[
  {
    "id": 2,
    "userId": 1,
    "code": 16372996967936,
    "name": "Comprehensive Test",
    "description": "",
    "createTime": "2025-01-20 11:40:39",
    "updateTime": "2025-01-20 11:40:39",
    "perm": 0,
    "defCount": 0,
    "instRunningCount": 0
  }
]

A pasta projects armazena definições de fluxo de trabalho. Na construção manual, altere o diretório de próximo nível para o nome do projeto. Em seguida, exporte as definições de fluxo de trabalho pela interface do DolphinScheduler, renomeie-as sequencialmente para process_definitions_page_*.json e coloque-as no diretório processDefinition.

Na página de lista de definições de fluxo de trabalho, clique em Export na parte superior da página para exportar o arquivo de definição do fluxo de trabalho.

Os diretórios datasource, resource e udfFunction contêm informações de fontes de dados, arquivos de recursos e UDFs, respectivamente. Como a interface do DolphinScheduler não possui recurso de exportação para esses elementos, eles podem ser omitidos. Preencha datasource_page_1.json, resources.json e udf_function_page_1.json com um array vazio ([]). A omissão desses elementos tem um impacto menor nos detalhes da migração do fluxo de trabalho. Isso afeta o mapeamento de nós SQL associados a fontes de dados, o mapeamento de nós DataX em modo de modelo não personalizado associados a fontes de dados e a migração de relações de referência entre nós e recursos. Os nós afetados são criados normalmente no DataWorks, mas configure manualmente a vinculação de nós a fontes de dados e recursos no DataWorks.

5,3. E se o token for válido, mas alguns fluxos de trabalho estiverem ausentes na exportação?

Primeiro, verifique se o token tem permissões para o projeto.

Algumas versões menores das APIs do DolphinScheduler 1.x podem causar perda de dados durante a exportação. Use o relatório estatístico nos resultados da exportação para identificar e adicionar fluxos de trabalho ausentes.

2,1. Modelo de configuração de transformação

• Antes de usar o arquivo, remova os comentários do código JSON.

{
  "conf": {},
  "self": {
    "if.use.default.convert": false,
    "if.use.migrationx.before": false,
    "if.use.dataworks.newidea": true,
    "owner.map": [ // Owner mapping
      {
        "src": "1", // DolphinScheduler user ID
        "tgt": "202006995118212119" // DataWorks user ID
      }
    ],
    "conf": [
      {
        "nodes": "all", // Scope of the rule group
        "rule": {
          "settings": {
            // Convert DolphinScheduler Shell nodes to DataWorks Shell nodes
            "workflow.converter.shellNodeType": "DIDE_SHELL",
            // Convert unknown nodes to DataWorks virtual nodes by default
            "workflow.converter.target.unknownNodeTypeAs": "VIRTUAL",
            // Convert DolphinScheduler SQL nodes to corresponding DataWorks SQL or database nodes based on the data source type
            "workflow.converter.dolphinscheduler.sqlNodeTypeMapping": {
              "CLICKHOUSE": "CLICK_SQL",
              "HIVE": "ODPS_SQL",
              "STARROCKS": "StarRocks",
              "DORIS": "HOLOGRES_SQL",
              "MYSQL": "MYSQL",
              "REDSHIFT": "Redshift",
              "SQLSERVER": "SQLSERVER",
              "PRESTO": "EMR_PRESTO",
              "POSTGRESQL": "POSTGRESQL",
              "ORACLE": "Oracle",
              "ATHENA": "MYSQL"
            },
            // Mapping of DolphinScheduler and DataWorks data source names
            "workflow.converter.connection.mapping": {
              "mysqlDb1": "dataworks_mysqlDb1",
              "srDb1": "dataworks_srDb1"
            },
            // Main compute engine bound on DataWorks (EMR/MaxCompute/Hologres)
            "workflow.converter.target.engine.type": "EMR",
            // Convert DolphinScheduler Spark nodes to DataWorks MaxCompute Spark nodes
            "workflow.converter.sparkSubmitAs": "ODPS_SPARK",
            "workflow.converter.sparkVersion": "3.x",
          }
        }
      }
    ]
  },
  "schedule_datasource": {
    "name": "DsProject",
    "type": "DolphinScheduler"
  },
  "target_schedule_datasource": {}
}

2,2. Mapeamento de proprietários

O DolphinScheduler registra o proprietário de cada fluxo de trabalho. A ferramenta suporta o mapeamento de usuários do DolphinScheduler para usuários do DataWorks, preservando a propriedade em fluxos de trabalho e nós.

Obtenha o nome de usuário e o ID do DolphinScheduler na página User Management.

Em um workspace do DataWorks, adicione usuários como membros. Obtenha o ID do usuário no canto superior direito.

Também obtenha o ID na lista suspensa Owner na página do DataStudio.

2,3. Regras de transformação de nós

2,3.1. Escopo das regras

Defina o escopo das regras de transformação de nós. Para aplicar uma regra uniforme a todos os nós, configure "nodes": "all" e especifique as configurações. Geralmente, um grupo de regras "all" é suficiente.

• Antes de usar o arquivo, remova os comentários do código JSON.

{
  "conf": {},
  "self": {
    "conf": [
      {
        "nodes": "all", // The scope of the rule group is ALL. All nodes are transformed according to this rule.
        "rule": {
          "settings": {
            // Settings
          }
      }
    ]
  }
}

Se alguns nós exigirem regras separadas, especifique o escopo em nodes inserindo IDs ou nomes de tarefas, separados por vírgulas. Recomendamos o uso de IDs, pois nomes podem levar a correspondências incorretas. Expressões regulares também são suportadas para corresponder a nomes de nós. Configure um grupo de regras normal como regra padrão para os nós restantes.

• Antes de usar o arquivo, remova os comentários do código JSON.

{
  "conf": {},
  "self": {
    "conf": [
      {
        "nodes": "node1Name, node2Id", // The scope of the rule group is node1 and node2.
        "rule": {
          "settings": {
            // Settings 1
          }
        },
        {
        "nodes": "node3Name, node4Id", // The scope of the rule group is node3 and node4.
        "rule": {
          "settings": {
            // Settings 2
          }
        },
        {
        "nodes": "regexExpression", // Supports filtering node names with regular expressions.
        "rule": {
          "settings": {
            // Settings 3
          }
        },
        {
        "nodes": "normal", // Transformation rule for the remaining nodes.
        "rule": {
          "settings": {
            // Settings 4
          }
        }
        ]
  }
}

2,3.2. Regras de transformação

O DolphinScheduler 1.x, 2.x e 3.x suportam tipos diferentes de nós, portanto as soluções de transformação e os itens de configuração variam.

2,3.2.1. Configuração de transformação do DolphinScheduler 3.x

Atualmente, a ferramenta suporta a transformação dos seguintes tipos de nós do DolphinScheduler 3.x:

SHELL, SQL, PYTHON, DATAX, SQOOP, SEATUNNEL, HIVECLI, SPARK (Java, Python, Sql), MR, PROCEDURE, HTTP, CONDITIONS, SWITCH, DEPENDENT, SUB_PROCESS

Os tipos para os quais você pode configurar regras de mapeamento do DataWorks incluem:

• SHELL (workflow.converter.shellNodeType):

Recomendamos convertê-los para nós DIDE_SHELL, EMR_SHELL ou VIRTUAL.

• SQL (workflow.converter.dolphinscheduler.sqlNodeTypeMapping):

Converta-os para vários nós SQL ou de banco de dados.

• PROCEDURE (workflow.converter.dolphinscheduler.sqlNodeTypeMapping):

Converta-os para vários nós SQL ou de banco de dados.

• PYTHON (workflow.converter.pyNodeType):

Recomendamos convertê-los para nós PYTHON, PYODPS, PYODPS3 ou EMR_SHELL.

• HIVECLI (workflow.converter.dolphinscheduler.sqlNodeTypeMapping/HIVE):

Converta-os para EMR_HIVE ou ODPS_SQL.

• SPARK (workflow.converter.sparkSubmitAs):

Recomendamos converter nós SparkJava e SparkPython para nós ODPS_SPARK ou EMR_SPARK.

Para nós SparkSql, converta-os para nós ODPS_SQL ou EMR_SPARK_SQL.

• MR (workflow.converter.mrNodeType):

Converta-os para nós ODPS_MR ou EMR_MR.

Para mais informações sobre os tipos de nós do DataWorks, consulte a seguinte classe de enumeração:

https://github.com/aliyun/dataworks-spec/blob/b0f4a4fd769215d5f81c0bbe990addd7498df5f4/spec/src/main/java/com/aliyun/dataworks/common/spec/domain/dw/types/CodeProgramType.java#L180

Tipos de nós com regras de transformação fixas:

• DATAX: Esses nós são convertidos para nós DI. A conversão suporta tanto o modo de modelo personalizado (modo JSON Script) quanto o modo regular (modo de preenchimento de formulário frontend).

As seguintes configurações de plugin de leitura de fonte de dados são suportadas: MYSQL -> mysql, POSTGRESQL -> postgresql, ORACLE -> oracle, SQLSERVER -> sqlserver, ODPS -> odps, OSS -> oss, HIVE -> hdfs, HDFS -> hdfs, CLICKHOUSE -> clickhouse, MONGODB -> mongodb.

As seguintes configurações de plugin de escrita de fonte de dados são suportadas: MYSQL -> mysql, POSTGRESQL -> postgresql, ORACLE -> oracle, SQLSERVER -> sqlserver, ODPS -> odps, OSS -> oss, HIVE -> hdfs, HDFS -> hdfs, CLICKHOUSE -> clickhouse, MONGODB -> mongodb.

• SQOOP: Estes nós são convertidos para nós DI.

Configurações de plugin de leitura de fonte de dados suportadas: Mysql -> mysql, Hive -> hive, HDFS -> hdfs.

Configurações de plugin de escrita de fonte de dados suportadas: Mysql -> mysql, Hive -> hive, HDFS -> hdfs.

• SEATUNNEL: Transforma um componente em um nó DI.

A transformação de script ainda não é suportada. Apenas o nó e suas informações de agendamento são transformados.

• HTTP: Tais nós são convertidos para nós DIDE_SHELL (Shell geral). A ferramenta de migração concatena automaticamente os parâmetros de solicitação em um comando curl.

• SWITCH: Estes nós são convertidos para nós CONTROLLER_BRANCH (ramificação). A funcionalidade permanece a mesma antes e depois da migração.

• SUB_PROCESS: Estes nós são convertidos para nós SUB_PROCESS. A funcionalidade permanece inalterada após a migração. Note que, ao importar para o DataWorks, a ferramenta de migração ativa a opção 'Can be referenced' para o fluxo de trabalho referenciado. O fluxo de trabalho referenciado só pode ser iniciado por uma chamada SUB_PROCESS e não pode ser agendado para execução independente.

• DEPENDENT: Estes nós são convertidos para nós VIRTUAL. A dependência é convertida em uma dependência de linhagem de nó. Por exemplo, se um nó Dependent depender do Fluxo de Trabalho A, a dependência é convertida em uma linhagem do nó final do Fluxo de Trabalho A para o nó Dependent. Se um nó Dependent depender do Nó A, a dependência é convertida em uma linhagem do Nó A para o nó Dependent. Os diagramas a seguir ilustram esses exemplos:

• CONDITIONS: Este nó contém duas camadas de lógica, implementadas usando um nó CONTROLLER_JOIN (junção) de duas camadas. Um nó CONDITIONS tem dois nós upstream, A e B, e dois nós downstream, C e D. A expressão lógica é ((!A&B)|(A&!B)|(!A&!B)). Se a expressão for verdadeira, o fluxo prossegue para C. Se for falsa, o fluxo prossegue para D. A camada superior gera três nós de junção para calcular os resultados de !A&B, A&!B e !A&!B. A camada inferior gera dois nós. Um nó aciona a execução do nó downstream C quando ((!A&B)|(A&!B)|(!A&!B))==true. O outro nó aciona a execução do nó downstream D quando (!(!A&B)&!(A&!B)&!(!A&!B))==true. Esse processo atinge o mesmo efeito do nó CONDITIONS original.

2,3.2.2. Configuração de transformação do DolphinScheduler 2.x

Atualmente, a ferramenta suporta a transformação dos seguintes tipos de nós do DolphinScheduler 2.x:

SHELL, SQL, PYTHON, DATAX, SQOOP, HIVECLI, SPARK (Java, Python, Sql), MR, PROCEDURE, HTTP, CONDITIONS, SWITCH, DEPENDENT, SUB_PROCESS

Em comparação com a versão 2.x, o DolphinScheduler 3.x adiciona apenas o tipo de nó SEATUNNEL. As soluções de transformação e os itens de configuração para os outros nós são os mesmos do DolphinScheduler 3.x. Para mais informações, consulte a seção anterior.

2,3.2.3. Configuração de transformação do DolphinScheduler 1.x

Atualmente, a ferramenta suporta a transformação dos seguintes tipos de nós do DolphinScheduler 1.x:

SHELL, SQL, PYTHON, DATAX, SQOOP, SPARK (Java, Python, Sql), MR, CONDITIONS, DEPENDENT, SUB_PROCESS

As soluções de transformação e os itens de configuração para esses nós são os mesmos do DolphinScheduler 3.x. Para mais informações, consulte a seção anterior.

sh ./bin/run.sh convert \
-c ./conf/<your_config_file>.JSON \
-f ./data/1_ReaderOutput/<source_discovery_export_package>.zip \
-o ./data/2_ConverterOutput/<transformation_result_package>.zip \
-t <PluginName>

sh ./bin/run.sh convert \
-c ./conf/projectA_convert.JSON \
-f ./data/1_ReaderOutput/projectA_ReaderOutput.zip \
-o ./data/2_ConverterOutput/projectA_ConverterOutput.zip \
-t dolphinv3-dw-converter

1,1. Transformação bem-sucedida

A ferramenta de transformação foi executada com êxito, as informações de agendamento de origem foram convertidas para o formato de agendamento do DataWorks e o arquivo ConverterOutput.zip foi gerado.

(Opcional, mas recomendado) Abra o pacote de saída da transformação e verifique o relatório estatístico para confirmar que o escopo da migração foi transformado com sucesso.

1,2. Configuração do DataWorks

No DataWorks, execute as seguintes ações:

1. Crie um workspace.

2. Crie um par de AccessKey e garanta que ele tenha permissões administrativas para o workspace. Recomendamos fortemente criar um par de AccessKey vinculado à sua conta para facilitar a solução de problemas caso ocorram erros de escrita.

3. No workspace, crie fontes de dados, vincule recursos de computação e crie grupos de recursos.

4. No workspace, carregue arquivos de recursos e crie UDFs.

1,3. Verificação de conectividade de rede

Verifique se é possível conectar-se ao endpoint do DataWorks.

Lista de endpoints:

Endpoints

ping dataworks.aliyuncs.com

{
  "schedule_datasource": {
    "name": "YourDataWorks", // Name your DataWorks data source.
    "type": "DataWorks",
    "properties": {
      "endpoint": "dataworks.cn-hangzhou.aliyuncs.com", // Endpoint
      "project_id": "YourProjectId", // Workspace ID
      "project_name": "YourProject", // Workspace name
      "ak": "************", // AK
      "sk": "************", // SK
    },
    "operaterType": "MANUAL"
  },
  "conf": {
    "di.resource.group.identifier": "Serverless_res_group_***_***", // Data integration resource group
    "resource.group.identifier": "Serverless_res_group_***_***", // Scheduling resource group
    "dataworks.node.type.xls": "/Software/bwm-client/conf/CodeProgramType.xls", // Path to the DataWorks node type table
    "qps.limit": 5 // QPS limit for sending API requests to DataWorks
  }
}

2,1. Endpoint

Selecione o endpoint com base na região onde seu workspace do DataWorks está localizado. Para mais informações, consulte:

Endpoints

2,2. ID e nome do workspace

Abra o console do DataWorks. Acesse a página do produto workspace e obtenha o ID e o nome do workspace nas informações básicas no lado direito da página.

2,3. Criar e autorizar um par de AccessKey

Na sua página de usuário, crie um par de AccessKey. O par de AccessKey deve ter permissões administrativas de leitura e escrita para o workspace do DataWorks de destino.

Faça login no console do Alibaba Cloud. Clique em sua foto de perfil no canto superior direito e acesse a página AccessKey Management. Clique em Create AccessKey. Após a criação da chave, registre o AccessKey ID e o AccessKey Secret.

O gerenciamento de permissões ocorre em dois locais. Se a conta for um usuário RAM, conceda primeiro a autorização ao usuário RAM para operações do DataWorks.

Página de políticas: https://ram.console.alibabacloud.com/policies

Na página Create policy, selecione a aba Visual editor. Defina Product como DataWorks, Actions como All Operations e Resources como All Resources. Clique em OK para criar a política.

Após a criação da política, acesse a página de detalhes da política. Na aba Authorization Management, clique em Add Authorization para conceder a política ao usuário RAM de destino.

Em seguida, no workspace do DataWorks, conceda permissões de workspace à conta.

Na página Workspace Members and Roles, adicione a conta de destino e defina sua função como Workspace Admin.

Nota: Defina uma política de restrição de acesso à rede para o AccessKey. Certifique-se de que o endereço IP da máquina onde a ferramenta de migração está sendo executada tenha acesso permitido.

2,4. Grupos de recursos

No painel de navegação à esquerda da página do produto workspace do DataWorks, escolha Resource Groups. Na página Resource Groups, anexe um grupo de recursos e obtenha seu ID.

Um grupo de recursos de uso geral pode ser usado tanto para agendamento de nós quanto para integração de dados. Configure o grupo de recursos de agendamento (resource.group.identifier) e o grupo de recursos de integração de dados (di.resource.group.identifier) para serem o mesmo grupo de recursos de uso geral.

2,5. Configurações de QPS

A ferramenta importa dados chamando APIs do DataWorks. Diferentes edições do DataWorks possuem limites distintos de consultas por segundo (QPS) e limites diários de chamadas para operações de leitura e escrita da OpenAPI. Para mais informações, consulte Limites.

Para as edições Basic, Standard e Professional do DataWorks, recomendamos definir "qps.limit" como 5. Para a edição Enterprise, recomendamos definir "qps.limit" como 20.

Nota: Evite executar várias ferramentas de importação simultaneamente.

2,6. Configurações de ID de tipo de nó do DataWorks

No DataWorks, alguns tipos de nós recebem TypeIDs diferentes em regiões distintas. O TypeID específico depende da interface do DataWorks DataStudio. Essa característica aplica-se principalmente a nós de banco de dados. Para mais informações, consulte Nós de banco de dados.

Por exemplo, o NodeTypeId para um nó MySQL é 1000039 na região China (Hangzhou) e 1000041 na região China (Shenzhen).

Para adaptar-se a essas diferenças regionais no DataWorks, a ferramenta fornece uma maneira configurável de especificar a tabela de TypeID de nó que ela utiliza.

Esta tabela inclui colunas como Node Type, Node Name, **programType (identificador de tipo de nó) e Is Active**. Ela lista todos os tipos de nós suportados e seus mapeamentos de ID correspondentes. Modifique os valores de acordo com os IDs reais na região de destino.

Especifique a tabela usando um item de configuração na ferramenta de importação:

"conf": {
    "dataworks.node.type.xls": "/Software/bwm-client/conf/CodeProgramType.xls" // Path to the DataWorks node type table
 }

Para obter o ID do tipo de nó na interface do DataWorks DataStudio, crie um fluxo de trabalho, crie um nó nesse fluxo de trabalho e clique em Save. Depois disso, visualize o Spec do fluxo de trabalho.

Após salvar, expanda Workflow Spec Definition no painel lateral direito. No conteúdo JSON, localize o valor do campo programType, que corresponde ao programType daquele nó.

Se o tipo de nó estiver configurado incorretamente, o seguinte erro será reportado ao publicar o fluxo de trabalho.

Exemplo de mensagem de erro: The program type of the set node #1303 does not exist, onde #1303 é o programType configurado incorretamente.

sh ./bin/run.sh write \
-c ./conf/<your_config_file>.json \
-f ./data/2_ConverterOutput/<conversion_result_output_package>.zip \
-o ./data/4_WriterOutput/<import_result_storage_package>.zip \
-t dw-newide-writer

sh ./bin/run.sh write \
-c ./conf/projectA_write.json \
-f ./data/2_ConverterOutput/projectA_ConverterOutput.zip \
-o ./data/4_WriterOutput/projectA_WriterOutput.zip \
-t dw-newide-writer