### Build Project with Maven

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/dev/develop-guide.md

Builds the project using Maven, skipping tests during the clean install and package phases. Requires Maven, Java, and Python.

```shell
mvn clean install -DskipTests=true && mvn package -DskipTests=true
```

--------------------------------

### MigrationX Reader Usage Example (DolphinScheduler)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Demonstrates how to use the MigrationX reader tool to export workflow configurations from DolphinScheduler. It specifies the application type and necessary API parameters.

```shell
bin/reader.py \
-a dolphinscheduler \
-e http://dolphinscheduler_api_host \
-t {token} \
-v 1.3.9  \
-p project_a \
-f project_a.zip
```

--------------------------------

### DolphinScheduler to DataWorks Conversion Example (Shell)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

An example of how to use the Transformer tool to convert a DolphinScheduler zip package to a DataWorks zip package. This involves specifying the transformer app, configuration file, source zip, and target zip.

```shell
bin/transformer.py \
-a dolphinscheduler_to_dataworks\
-c dataworks-transformer-config.json\
-s project_a.zip\
-t dw.zip
```

--------------------------------

### MigrationX Reader Usage Example (Airflow)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Illustrates the usage of the MigrationX reader tool for exporting Airflow DAGs. It requires the path to the DAG folder and an output file for the JSON configuration.

```shell
bin/reader.py \
-a airflow \
-d /path/to/dag_folder \
-o output.json
```

--------------------------------

### Branch and Join Nodes Example (Java)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Demonstrates how to implement conditional logic in FlowSpec using branch nodes, which execute different paths based on specified conditions. It also shows how join nodes can merge these execution paths.

```java
String branchSpec = """
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "spec": {
    "artifacts": {
      "outputs": [
        {"id": "branch_1", "type": "output", "data": "autotest.branch_1"},
        {"id": "branch_2", "type": "output", "data": "autotest.branch_2"}
      ]
    },
    "nodes": [
      {
        "id": "branch_node",
        "branch": {
          "branches": [
            {
              "when": "a == 1",
              "output": "{{artifacts.branch_1}}",
              "desc": "First branch condition"
            },
            {
              "when": "a == 2",
              "output": "{{artifacts.branch_2}}",
              "desc": "Second branch condition"
            }
          ]
        }
      },
      {
        "id": "task_branch_1",
        "script": {"runtime": {"command": "ODPS_SQL"}}
      },
      {
        "id": "task_branch_2",
        "script": {"runtime": {"command": "DIDE_SHELL"}}
      }
    ],
    "flow": [
      {
        "nodeId": "{{task_branch_1}}",
        "depends": [{"nodeId": "{{branch_node}}", "output": "{{artifacts.branch_1}}"}]
      },
      {
        "nodeId": "{{task_branch_2}}",
        "depends": [{"nodeId": "{{branch_node}}", "output": "{{artifacts.branch_2}}"}]
      }
    ]
  }
}
"";

Specification<DataWorksWorkflowSpec> branchWorkflow = SpecUtil.parseToDomain(branchSpec);
SpecNode branchNode = branchWorkflow.getSpec().getNodes().stream()
    .filter(n -> n.getBranch() != null)
    .findFirst().orElseThrow();

branchNode.getBranch().getBranches().forEach(branch ->
    System.out.println("Branch: " + branch.getWhen() + " -> " + branch.getDesc()));

```

--------------------------------

### MigrationX Reader Usage Example (Aliyun EMR)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Shows how to use the MigrationX reader tool to extract workflow configurations from Aliyun EMR. This includes EMR endpoint, access credentials, region, and project names.

```shell
bin/reader.py \
-a aliyunemr \
-d ./target/dump \
-e emr.aliyuncs.com \
-i ${accessId} \
-k ${accessKey} \
-r ${regionId} \
-p emr_prj01,emr_prj02
```

--------------------------------

### Setup Environment Variables for DolphinScheduler to DataWorks Transformation

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Configures environment variables required for the MigrationX tool to migrate workflows from DolphinScheduler to DataWorks. This includes DolphinScheduler API details and Aliyun DataWorks credentials.

```shell
# dolphinscheduler config variables
export DOLPHINSCHEDULER_API_ENDPOINT=http://server:port
export DOLPHINSCHEDULER_API_TOKEN=md5_token_get_from_dolphinscheduler 
export DOLPHINSCHEDULER_VERSION=1.3.9
export DOLPHINSCHEDULER_PROJECT_NAME=your_dolphinscheduler_project_name

# aliyun config variables
export ALIYUN_ACCESS_KEY_ID=your_access_key_id
export ALIYUN_ACCESS_KEY_SECRET=your_access_key_secret
export ALIYUN_REGION_ID=cn-shanghai
# dataworks workspace id to be imported
export ALIYUN_DATAWORKS_WORKSPACE_ID=your_dataworks_workspace_id
```

--------------------------------

### Set Project Version with Maven

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/dev/develop-guide.md

Increases the project version number consistently using the Maven versions plugin. Requires Maven to be installed.

```shell
mvn versions:set -DnewVersion={major.minor.patch}
```

--------------------------------

### Setup Environment Variables for Airflow to DataWorks Transformation

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Configures necessary environment variables for the AirWorks tool to transform Airflow DAGs into DataWorks workflows. This includes setting Python home, Airflow DAG folder, and DataWorks authentication credentials.

```shell
# environment variables for transforming Airflow DAGs
export PYTHON_HOME=/path/to/python/home/which/is/used/by/airflow
export PATH=${PYTHON_HOME}/bin:${PATH}
export AIRFLOW_V2_DAG_FOLDER=/path/to/airflow/dags/floder/
export FLOWSPECT_FOLDER_FOR_AIRFLOW_DAGS=/path/to/folder/for/saving/exported/flowspecs/
export MIGRATIONX_HOME=/path/to/migration/tool/root/folder/

# environment variables fro commiting to DataWorks
export ALIYUN_REGION_ID=dataworks-located-region(e.g. cn-shanghai)
export ALIYUN_ACCESS_KEY_ID=your_access_key_id
export ALIYUN_ACCESS_KEY_SECRET=your_access_key_secret
export ALIYUN_DATAWORKS_WORKSPACE_ID=your_dataworks_workspace_id
```

--------------------------------

### ForEach Loop Node Configuration in FlowSpec

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Defines a ForEach loop node in FlowSpec, iterating over an array and executing inner nodes for each element. It includes variable definition and an example of accessing loop properties.

```java
String foreachSpec = """
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "spec": {
    "variables": [
      {
        "id": "var_arr",
        "name": "var_arr",
        "type": "System",
        "scope": "NodeContext",
        "value": "['a','b','c']"
      }
    ],
    "nodes": [
      {
        "id": "foreach_node",
        "for-each": {
          "array": "{{variables.var_arr}}",
          "nodes": [
            {
              "id": "inner_sql",
              "script": {
                "path": "/sql.sql",
                "runtime": {"engine": "MaxCompute", "command": "ODPS_SQL"}
              }
            },
            {
              "id": "inner_shell",
              "script": {
                "path": "/shell.sh",
                "runtime": {"command": "DIDE_SHELL"}
              }
            },
            {"id": "end"}
          ],
          "flow": [
            {
              "nodeId": "{{end}}",
              "depends": [{"nodeId": "{{inner_sql}}"}, {"nodeId": "{{inner_shell}}"}]
            }
          ]
        }
      }
    ]
  }
}
"""

Specification<DataWorksWorkflowSpec> foreachWorkflow = SpecUtil.parseToDomain(foreachSpec);
SpecNode foreachNode = foreachWorkflow.getSpec().getNodes().get(0);

System.out.println("ForEach array: " + foreachNode.getForeach().getArray().getValue());
System.out.println("Inner nodes count: " + foreachNode.getForeach().getNodes().size());
```

--------------------------------

### Command-Line Workflow Migration

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Illustrates the command-line approach for migrating workflows using a reader, transformer, and writer pipeline. This method is suitable for batch migrations between different scheduling systems.

```bash
# Example command structure (actual command will vary based on specific tools)
reader --input <source_workflow_file> | transformer --config <migration_config> | writer --output <target_workflow_file>
```

--------------------------------

### Transformer CLI Usage (Shell)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Defines the command-line options for the Transformer tool, used to convert workflow configurations. It accepts arguments for the transformer app, configuration file, source package, and target package.

```shell
usage: Options
Transformer Command App
 -a,--app <arg>             transformer app name, dolphinscheduler_to_dataworks etc.   
 -c,--config <arg>          transform configuration file path
 -s,--sourcePackage <arg>   source package file path
 -t,--targetPackage <arg>   target package file path
```

--------------------------------

### Java API for Spec Parsing and Writing

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Demonstrates the direct Java API usage for parsing workflow specifications into domain objects and writing domain objects back to a specification format. This is useful for programmatic manipulation of workflows.

```java
import com.aliyun.dataworks.common.spec.SpecUtil;

// ... in a Java class ...

// Parse spec to domain object
Object domainObject = SpecUtil.parseToDomain("path/to/spec.json");

// Write domain object to spec
SpecUtil.writeToSpec(domainObject, "path/to/output.yaml");
```

--------------------------------

### End-to-End DolphinScheduler to DataWorks Migration

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This command executes a complete migration from DolphinScheduler to DataWorks by combining read, transform, and write operations. Environment variables for both DolphinScheduler and Aliyun must be set.

```bash
# Configure environment
export DOLPHINSCHEDULER_API_ENDPOINT=http://dolphinscheduler:12345
export DOLPHINSCHEDULER_API_TOKEN=your_token
export DOLPHINSCHEDULER_VERSION=1.3.9
export DOLPHINSCHEDULER_PROJECT_NAME=my_project
export ALIYUN_ACCESS_KEY_ID=your_access_key_id
export ALIYUN_ACCESS_KEY_SECRET=your_access_key_secret
export ALIYUN_REGION_ID=cn-shanghai
export ALIYUN_DATAWORKS_WORKSPACE_ID=your_workspace_id

# Run complete migration
bin/migrationx.py
```

--------------------------------

### File Resources and Functions Definition in FlowSpec

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Demonstrates how to define file resources (like JARs) and user-defined functions (UDFs) within FlowSpec. It shows accessing these defined resources and functions programmatically.

```java
String resourceSpec = """
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "spec": {
    "fileResources": [
      {
        "id": "udf_jar",
        "name": "geo_udf.jar",
        "type": "jar",
        "file": {
          "storage": {"type": "oss"},
          "path": "/path/to/geo_udf.jar",
          "extension": ".jar"
        }
      }
    ],
    "functions": [
      {
        "id": "geo_loc",
        "name": "geo_loc",
        "type": "MATH",
        "className": "com.aliyun.odps.udf.example.GeoLocation",
        "datasource": {"type": "odps", "name": "autotest"},
        "runtimeResource": {"resourceGroup": "group1"},
        "fileResources": ["{{udf_jar}}"],
        "usageDescription": "Calculate geographic location from IP address",
        "usageExample": "SELECT geo_loc('127.0.0.1', city, province) FROM table1;"
      }
    ]
  }
}
"""

Specification<DataWorksWorkflowSpec> resourceWorkflow = SpecUtil.parseToDomain(resourceSpec);

// Access file resources
resourceWorkflow.getSpec().getFileResources().forEach(res ->
    System.out.println("Resource: " + res.getName() + " (" + res.getType() + ")"));

// Access functions
resourceWorkflow.getSpec().getFunctions().forEach(func ->
    System.out.println("Function: " + func.getName() + " - " + func.getClassName()));
```

--------------------------------

### Writer CLI Usage (Shell)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Defines the command-line options for the Writer tool, used to write workflow configurations to a target system like DataWorks. It requires specifying the app, DataWorks endpoint, access keys, workspace ID, region, and the input package file.

```shell
usage: Options
migrationx
 -a,--app <arg>   app name

Available apps:
 {
  "writer": [
    "dataworks"
  ]
}
```

--------------------------------

### Complete Workflow with Variables and Dependencies (Java)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Defines a complete workflow using FlowSpec, including variables, triggers, runtime resources, and script nodes with dependencies. It demonstrates how to parse the workflow specification and access its components like variables and flow dependencies.

```java
String workflowSpec = """
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "spec": {
    "variables": [
      {
        "id": "bizdate",
        "name": "bizdate",
        "type": "System",
        "scope": "Workflow",
        "value": "$bizdate"
      },
      {
        "id": "ds",
        "name": "ds",
        "scope": "NodeParameter",
        "type": "System",
        "value": "$[yyyymmddhh24mi-30/24/60]"
      }
    ],
    "triggers": [
      {
        "id": "every_30_minutes",
        "type": "Scheduler",
        "cron": "00 */30 00-23 * * ?",
        "startTime": "2023-01-01T00:00:00",
        "endTime": "2024-01-01T00:00:00"
      }
    ],
    "runtimeResources": [
      {
        "id": "default_resgroup",
        "resourceGroup": "S_resgroup_xxx"
      }
    ],
    "scripts": [
      {
        "id": "di_script",
        "path": "/DataIntegration/sync_job",
        "runtime": {
          "engine": "DataIntegration",
          "command": "DI"
        },
        "parameters": ["{{variables.bizdate}}"]
      },
      {
        "id": "sql_script",
        "path": "/MaxCompute/DataAnalytic/process",
        "language": "sql",
        "runtime": {
          "engine": "MaxCompute",
          "command": "ODPS_SQL"
        },
        "parameters": ["{{variables.bizdate}}", "{{variables.ds}}"]
      }
    ],
    "nodes": [
      {
        "id": "data_sync",
        "name": "data_sync",
        "script": "{{scripts.di_script}}",
        "trigger": "{{triggers.every_30_minutes}}",
        "runtimeResource": "{{runtimeResources.default_resgroup}}",
        "recurrence": "Normal",
        "priority": 7,
        "timeout": 4
      },
      {
        "id": "sql_process",
        "name": "sql_process",
        "script": "{{scripts.sql_script}}",
        "trigger": "{{triggers.every_30_minutes}}",
        "runtimeResource": "{{runtimeResources.default_resgroup}}"
      }
    ],
    "flow": [
      {
        "nodeId": "{{sql_process}}",
        "depends": [
          {
            "nodeId": "{{data_sync}}",
            "type": "Normal"
          }
        ]
      }
    ]
  }
}
"";

Specification<DataWorksWorkflowSpec> spec = SpecUtil.parseToDomain(workflowSpec);
DataWorksWorkflowSpec workflow = spec.getSpec();

// Access variables
workflow.getVariables().forEach(var ->
    System.out.println("Variable: " + var.getName() + " = " + var.getValue()));

// Access flow dependencies
workflow.getFlow().forEach(flowDepend -> {
    System.out.println("Node: " + flowDepend.getNodeId().getId());
    flowDepend.getDepends().forEach(dep ->
        System.out.println("  Depends on: " + dep.getNodeId().getId()));
});

```

--------------------------------

### End-to-End Migration Script (Python)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Showcases the single-command execution of the `migrationx.py` script for streamlined migration from systems like DolphinScheduler or Airflow to DataWorks. This provides an end-to-end solution for common migration scenarios.

```python
# Example command for migrating from Airflow to DataWorks
python migrationx.py --source airflow --target dataworks --input_dir <airflow_dag_directory> --output_dir <dataworks_spec_directory>
```

--------------------------------

### Export Aliyun EMR Workflows using reader.py

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This command exports Aliyun EMR workflows to a specified target directory. It requires Aliyun credentials, region, and project details. The output is typically a zip archive.

```bash
bin/reader.py \
  -a aliyunemr \
  -d ./target/dump \
  -e emr.aliyuncs.com \
  -i ${ALIYUN_ACCESS_KEY_ID} \
  -k ${ALIYUN_ACCESS_KEY_SECRET} \
  -r cn-shanghai \
  -p emr_project_01,emr_project_02
```

--------------------------------

### Run MigrationX for DolphinScheduler to DataWorks Transformation

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Executes the MigrationX script to perform the transformation of DolphinScheduler workflows to DataWorks. This command initiates the migration process using the configured settings.

```shell
bin/migrationx.py
```

--------------------------------

### DataWorksNodeAdapter for Node Context Handling (Java)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This Java code demonstrates how to use the DataWorksNodeAdapter to access resolved inputs and outputs for a DataWorks node within a workflow specification. It utilizes the `SpecUtil` for parsing and `Context` builder for deployment settings.

```java
import com.aliyun.dataworks.common.spec.domain.dw.nodemodel.DataWorksNodeAdapter;
import com.aliyun.dataworks.common.spec.domain.dw.nodemodel.DataWorksNodeAdapter.Context;

// Parse specification
Specification<DataWorksWorkflowSpec> spec = SpecUtil.parseToDomain(workflowJson);
SpecNode node = spec.getSpec().getNodes().get(0);

// Create adapter with deployment context
DataWorksNodeAdapter adapter = new DataWorksNodeAdapter(
    spec,
    node,
    Context.builder()
        .deployToScheduler(true)
        .build()
);

// Access resolved inputs (including dependencies from flow)
List<Input> inputs = adapter.getInputs();
inputs.forEach(input -> {
    if (input instanceof SpecNodeOutput) {
        System.out.println("Input dependency: " + ((SpecNodeOutput) input).getData());
    }
});

// Access outputs
List<Output> outputs = adapter.getOutputs();
outputs.forEach(output -> {
    if (output instanceof SpecNodeOutput) {
        System.out.println("Output: " + ((SpecNodeOutput) output).getData());
    }
});
```

--------------------------------

### MigrationX Configuration for DolphinScheduler to DataWorks

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Defines the MigrationX configuration in JSON format for transforming DolphinScheduler workflows to DataWorks. It specifies the reader (DolphinScheduler), transformer, and writer (DataWorks) components and their parameters.

```json
{
  "reader": {
    "name": "dolphinscheduler",
    "params": [
      "-a dolphinscheduler",
      "-e ${DOLPHINSCHEDULER_API_ENDPOINT}",
      "-t ${DOLPHINSCHEDULER_API_TOKEN}",
      "-v ${DOLPHINSCHEDULER_VERSION}",
      "-p ${DOLPHINSCHEDULER_PROJECT_NAME}",
      "-f ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}.zip"
    ]
  },
  "transformer": {
    "name": "dolphinscheduler_to_dataworks",
    "params": [
      "-a dolphinscheduler_to_dataworks",
      "-c ${MIGRATIONX_HOME}/conf/dataworks-transformer-config.json",
      "-s ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}.zip",
      "-t ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}_dw.zip"
    ]
  },
  "writer": {
    "name": "dataworks",
    "params": [
      "-a dataworks",
      "-e dataworks.${ALIYUN_REGION_ID}.aliyuncs.com",
      "-i ${ALIYUN_ACCESS_KEY_ID}",
      "-k ${ALIYUN_ACCESS_KEY_SECRET}",
      "-p ${ALIYUN_DATAWORKS_WORKSPACE_ID}",
      "-r ${ALIYUN_REGION_ID}",
      "-f ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}_dw.zip",
      "-t SPEC"
    ]
  }
}
```

--------------------------------

### Parse and Serialize FlowSpec with SpecUtil in Java

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Demonstrates how to use the `SpecUtil` class in Java to parse a FlowSpec JSON string into domain objects and then serialize those domain objects back into a JSON spec string. This API is crucial for workflow manipulation and transformation within the DataWorks ecosystem.

```java
import com.aliyun.dataworks.common.spec.SpecUtil;
import com.aliyun.dataworks.common.spec.domain.DataWorksWorkflowSpec;
import com.aliyun.dataworks.common.spec.domain.Specification;
import com.aliyun.dataworks.common.spec.domain.ref.SpecNode;

// Parse FlowSpec JSON to domain object
String specJson = """
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "metadata": {
    "owner": "26730",
    "description": "Example workflow"
  },
  "spec": {
    "nodes": [
      {
        "id": "node_1",
        "name": "sql_task",
        "script": {
          "path": "/DataAnalytic/sql_task.sql",
          "language": "sql",
          "runtime": {
            "engine": "MaxCompute",
            "command": "ODPS_SQL"
          }
        },
        "trigger": {
          "type": "Scheduler",
          "cron": "00 00 00 * * ?",
          "startTime": "2023-01-01T00:00:00",
          "endTime": "2024-12-31T00:00:00"
        }
      }
    ],
    "flow": [
      {
        "nodeId": "node_1",
        "depends": []
      }
    ]
  }
}
""";

Specification<DataWorksWorkflowSpec> specification = SpecUtil.parseToDomain(specJson);

// Access parsed domain objects
DataWorksWorkflowSpec workflowSpec = specification.getSpec();
List<SpecNode> nodes = workflowSpec.getNodes();
SpecNode node = nodes.get(0);

System.out.println("Workflow Kind: " + specification.getKind());       // CycleWorkflow
System.out.println("Node Name: " + node.getName());                    // sql_task
System.out.println("Runtime Command: " + node.getScript().getRuntime().getCommand());  // ODPS_SQL

// Serialize domain object back to JSON spec
String outputJson = SpecUtil.writeToSpec(specification);
System.out.println(outputJson);
```

--------------------------------

### MigrationX Complete Migration Configuration

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This JSON configuration file defines the parameters for the reader, transformer, and writer components used in an end-to-end migration. It specifies source endpoints, transformation settings, and DataWorks import details.

```json
// conf/migrationx.json - Complete migration configuration
{
  "reader": {
    "name": "dolphinscheduler",
    "params": [
      "-a dolphinscheduler",
      "-e ${DOLPHINSCHEDULER_API_ENDPOINT}",
      "-t ${DOLPHINSCHEDULER_API_TOKEN}",
      "-v ${DOLPHINSCHEDULER_VERSION}",
      "-p ${DOLPHINSCHEDULER_PROJECT_NAME}",
      "-f ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}.zip"
    ]
  },
  "transformer": {
    "name": "dolphinscheduler_to_dataworks",
    "params": [
      "-a dolphinscheduler_to_dataworks",
      "-c ${MIGRATIONX_HOME}/conf/dataworks-transformer-config.json",
      "-s ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}.zip",
      "-t ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}_dw.zip"
    ]
  },
  "writer": {
    "name": "dataworks",
    "params": [
      "-a dataworks",
      "-e dataworks.${ALIYUN_REGION_ID}.aliyuncs.com",
      "-i ${ALIYUN_ACCESS_KEY_ID}",
      "-k ${ALIYUN_ACCESS_KEY_SECRET}",
      "-p ${ALIYUN_DATAWORKS_WORKSPACE_ID}",
      "-r ${ALIYUN_REGION_ID}",
      "-f ${PWD}/${DOLPHINSCHEDULER_PROJECT_NAME}_dw.zip",
      "-t SPEC"
    ]
  }
}
```

--------------------------------

### Run AirWorks for Airflow DAGs to DataWorks Transformation

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

Executes the AirWorks script to initiate the transformation of Airflow DAGs into DataWorks workflows. This command assumes that all necessary environment variables and configurations are set.

```shell
bin/airworks.py
```

--------------------------------

### DataWorks Transformer Configuration (JSON)

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage.md

A sample JSON configuration file for the DataWorks Transformer. This configuration specifies how different node types from the source workflow system should be mapped to DataWorks or EMR-related nodes.

```json
{
  "format": "SPEC",
  "locale": "zh_CN",
  "settings": {
    "workflow.converter.shellNodeType": "EMR_SHELL",
    "workflow.converter.commandSqlAs": "EMR_HIVE",
    "workflow.converter.sparkSubmitAs": "EMR_SPARK",
    "workflow.converter.target.unknownNodeTypeAs": "DIDE_SHELL",
    "workflow.converter.mrNodeType": "EMR_MR",
    "workflow.converter.target.engine.type": "EMR",
    "workflow.converter.dolphinscheduler.sqlNodeTypeMapping": {
      "POSTGRESQL": "EMR_HIVE",
      "MYSQL": "EMR_HIVE",
      "HIVE": "EMR_HIVE"
    }
  }
}
```

--------------------------------

### Import FlowSpec Package to DataWorks using writer.py

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This command imports a transformed FlowSpec package (zip file) into DataWorks using the writer.py script. It requires Aliyun credentials, workspace ID, and region information.

```bash
# Set Aliyun credentials
export ALIYUN_ACCESS_KEY_ID=your_access_key_id
export ALIYUN_ACCESS_KEY_SECRET=your_access_key_secret
export ALIYUN_REGION_ID=cn-shanghai
export ALIYUN_DATAWORKS_WORKSPACE_ID=your_workspace_id

# Import FlowSpec package to DataWorks
bin/writer.py \
  -a dataworks \
  -e dataworks.${ALIYUN_REGION_ID}.aliyuncs.com \
  -i ${ALIYUN_ACCESS_KEY_ID} \
  -k ${ALIYUN_ACCESS_KEY_SECRET} \
  -p ${ALIYUN_DATAWORKS_WORKSPACE_ID} \
  -r ${ALIYUN_REGION_ID} \
  -f my_project_dataworks.zip \
  -t SPEC
```

--------------------------------

### Transform DolphinScheduler Export to DataWorks Format

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This command uses the transformer.py script to convert a DolphinScheduler workflow export (zip file) into the DataWorks FlowSpec format. A configuration file is required to specify transformation settings.

```bash
bin/transformer.py \
  -a dolphinscheduler_to_dataworks \
  -c conf/dataworks-transformer-config.json \
  -s my_project.zip \
  -t my_project_dataworks.zip
```

--------------------------------

### MigrationX: Export DolphinScheduler Workflows (Bash)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Uses the MigrationX reader tool to export workflow configurations from DolphinScheduler. This involves setting necessary environment variables for API endpoint, token, and version, then executing the export command to a zip file.

```bash
# Set environment variables
export DOLPHINSCHEDULER_API_ENDPOINT=http://dolphinscheduler-server:12345
export DOLPHINSCHEDULER_API_TOKEN=your_api_token_here
export DOLPHINSCHEDULER_VERSION=1.3.9
export DOLPHINSCHEDULER_PROJECT_NAME=my_project

# Export DolphinScheduler workflows to zip file
bin/reader.py \
  -a dolphinscheduler \
  -e ${DOLPHINSCHEDULER_API_ENDPOINT} \
  -t ${DOLPHINSCHEDULER_API_TOKEN} \
  -v ${DOLPHINSCHEDULER_VERSION} \
  -p ${DOLPHINSCHEDULER_PROJECT_NAME} \
  -f ${DOLPHINSCHEDULER_PROJECT_NAME}.zip
```

--------------------------------

### Export Airflow DAGs using reader.py

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This command exports Airflow DAGs to a JSON file using the reader.py script. It requires the path to the Airflow DAGs directory.

```bash
bin/reader.py \
  -a airflow \
  -d /path/to/airflow/dags \
  -o airflow_export.json
```

--------------------------------

### Write to DataWorks using Shell Script

Source: https://github.com/aliyun/dataworks-spec/blob/master/docs/migrationx/usage_zh_CN.md

This shell script utilizes the writer.py utility to upload a DataWorks import package. It requires Alibaba Cloud credentials (access key ID and secret), workspace ID, and region. The script specifies the DataWorks API endpoint and the path to the zip file containing the data.

```shell
bin/writer.py \
 -a dataworks \
 -e dataworks.cn-shanghai.aliyuncs.com \
 -i $ALIYUN_ACCESS_KEY_ID \
 -k $ALIYUN_ACCESS_KEY_SECRET \
 -p $ALIYUN_DATAWORKS_WORKSPACE_ID \
 -r cn-shanghai \
 -f demo_space.zip
```

--------------------------------

### Spec Properties API

Source: https://github.com/aliyun/dataworks-spec/blob/master/schema/docs/flow-then-properties-spec.md

This section details the properties available within the DataWorks Spec schema, focusing on the 'spec' type and its associated properties like 'triggers'.

```APIDOC
## Spec Properties API

### Description
This API documentation describes the properties of the 'spec' type within the DataWorks flow schema. It details the structure and types of these properties, including nested objects and arrays.

### Method
GET

### Endpoint
/schemas/1.1.0/flow.schema.json#/then/properties/spec

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```json
{
  "example": "No request body needed for schema introspection."
}
```

### Response
#### Success Response (200)
- **triggers** (array) - An array of trigger objects defining the periodic scheduling. Each trigger object conforms to the [Trigger](trigger.md) schema.

#### Response Example
```json
{
  "spec": {
    "triggers": [
      {
        "type": "cron",
        "cronExpression": "0 0 * * * ?",
        "timezone": "Asia/Shanghai"
      }
    ]
  }
}
```

## Triggers Object Definition

### Description
This details the structure and properties of the 'triggers' array within the 'spec' object. It specifies that 'triggers' is an array of trigger objects, each defining a periodic scheduling configuration.

### Method
GET

### Endpoint
/schemas/1.1.0/flow.schema.json#/then/properties/spec/properties/triggers

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```json
{
  "example": "No request body needed for schema introspection."
}
```

### Response
#### Success Response (200)
- **triggers** (array) - An array of trigger objects. Each object has properties like `type`, `cronExpression`, `timezone`, etc., defining the schedule.

#### Response Example
```json
{
  "triggers": [
    {
      "type": "cron",
      "cronExpression": "0 0 * * * ?",
      "timezone": "Asia/Shanghai"
    }
  ]
}
```
```

--------------------------------

### DataWorks Transformer Configuration for EMR

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

This JSON configuration file defines settings for transforming EMR workflows into DataWorks format. It includes mappings for node types and engine configurations.

```json
// dataworks-transformer-config.json - EMR transformation configuration
{
  "format": "SPEC",
  "locale": "en_US",
  "settings": {
    "workflow.converter.shellNodeType": "EMR_SHELL",
    "workflow.converter.commandSqlAs": "EMR_HIVE",
    "workflow.converter.sparkSubmitAs": "EMR_SPARK",
    "workflow.converter.target.unknownNodeTypeAs": "DIDE_SHELL",
    "workflow.converter.mrNodeType": "EMR_MR",
    "workflow.converter.target.engine.type": "EMR",
    "workflow.converter.dolphinscheduler.sqlNodeTypeMapping": {
      "POSTGRESQL": "EMR_HIVE",
      "MYSQL": "EMR_HIVE",
      "HIVE": "EMR_HIVE"
    }
  }
}
```

--------------------------------

### EMR Hive Node Configuration in FlowSpec (YAML)

Source: https://context7.com/aliyun/dataworks-spec/llms.txt

Configures an EMR Hive node within FlowSpec using YAML format. It specifies job details, script content, and advanced runtime settings including Spark configurations and EMR job specific parameters.

```yaml
# EMR Hive node configuration (YAML format)
version: 1.1.0
kind: CycleWorkflow
spec:
  nodes:
    - id: emr_hive_node
      name: hive_task
      recurrence: Normal
      timeout: 3
      instanceMode: T+1
      rerunMode: Allowed
      rerunTimes: 3
      rerunInterval: 180000
      script:
        path: test/EMR_biz/EMR/hive_task
        language: hive-sql
        content: "SELECT * FROM source_table WHERE dt = '${bizdate}'"
        runtime:
          engine: Hive
          command: EMR_HIVE
          sparkConf:
            spark.executor.memory: 1024m
            spark.executor.cores: 1
            spark.executor.instances: 1
            spark.yarn.queue: default
          emrJobConfig:
            submitMode: Local
            submitter: root
            priority: 1
            queue: default
            cores: 1
            memory: 1024
            enableJdbcSql: true
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.