Este tema describe cómo usar Open API para crear, actualizar y eliminar tareas de Data Integration que sincronizan datos desde un origen hacia un destino.
Requisitos previos
Se creó un flujo de trabajo. Para obtener más información, consulte Create a scheduled workflow.
Se crearon los orígenes de datos obligatorios para la tarea de sincronización.
Limitaciones
Llame a la operación CreateDISyncTask para crear una tarea de Data Integration. El contenido de la tarea solo se configura en code editor.
No es posible usar Open API para crear un flujo de trabajo en DataWorks. En su lugar, utilice un flujo de trabajo existente para la tarea de sincronización de datos.
Obtención de un SDK
Obtenga el SDK más reciente desde el portal de Open API de Alibaba Cloud.
Procedimiento
Una vez configurado el entorno, siga estos pasos para crear y gestionar una tarea de sincronización de datos con la API de DataWorks:
Pasos
-
Cree una tarea de integración de datos.
Llame a la operación CreateDISyncTask para crear la tarea. El siguiente código de ejemplo muestra cómo configurar varios parámetros clave. Para obtener más información sobre los parámetros, consulte CreateDISyncTask.
public void createFile() throws ClientException{ CreateDISyncTaskRequest request = new CreateDISyncTaskRequest(); request.setProjectId(181565L); request.setTaskType("DI_OFFLINE"); request.setTaskContent("{\"type\":\"job\",\"version\":\"2.0\",\"steps\":[{\"stepType\":\"mysql\",\"parameter\":{\"envType\":1,\"datasource\":\"dh_mysql\",\"column\":[\"id\",\"name\"],\"tableComment\":\"Comment for the table same\",\"connection\":[{\"datasource\":\"dh_mysql\",\"table\":[\"same\"]}],\"where\":\"\",\"splitPk\":\"id\",\"encoding\":\"UTF-8\"},\"name\":\"Reader\",\"category\":\"reader\"},{\"stepType\":\"odps\",\"parameter\":{\"partition\":\"pt=${bizdate}\",\"truncate\":true,\"datasource\":\"odps_source\",\"envType\":1,\"column\":[\"id\",\"name\"],\"emptyAsNull\":false,\"tableComment\":\"Comment for the table same\",\"table\":\"same\"},\"name\":\"Writer\",\"category\":\"writer\"}],\"setting\":{\"errorLimit\":{\"record\":\"\"},\"speed\":{\"throttle\":false,\"concurrent\":2}},\"order\":{\"hops\":[{\"from\":\"Reader\",\"to\":\"Writer\"}]}}"); request.setTaskParam("{\"FileFolderPath\":\"Workflow/new_biz/Data Integration\",\"ResourceGroup\":\"S_res_group_280749521950784_1602767279794\"}"); request.setTaskName("new_di_task_0607_1416"); String regionId = "cn-hangzhou"; // Please ensure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.https://www.alibabacloud.com/help/zh/alibaba-cloud-sdk-262060/latest/configure-credentials-378659 IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")); DefaultProfile.addEndpoint("cn-hangzhou","dataworks-public","dataworks.cn-hangzhou.aliyuncs.com"); IAcsClient client; client = new DefaultAcsClient(profile); CreateDISyncTaskResponse response1 = client.getAcsResponse(request); Gson gson1 = new Gson(); System.out.println(gson1.toJson(response1)); } -
Llame a la operación UpdateFile para actualizar los parámetros de programación.
La siguiente tabla describe los parámetros de la solicitud.
Nombre
Tipo
Obligatorio
Ejemplo
Descripción
Action
String
Sí
UpdateFile
Operación que se va a ejecutar.
FileFolderPath
String
No
Workflows/MyFirstWorkflow/DataIntegration/Folder1/Folder2
Ruta del archivo.
ProjectId
Long
No
10000
ID del espacio de trabajo de DataWorks. Vaya a la página de administración de espacios de trabajo en la DataWorks Console para consultar el ID del espacio de trabajo.
FileName
String
No
ods_user_info_d
Nombre del archivo. Para cambiar el nombre de un archivo, establezca un valor nuevo en este parámetro.
Utilice la operación
FileIdpara obtener elListFilesnecesario.FileDescription
String
No
This is a file description
Descripción del archivo.
Content
String
No
SELECT "1";
Código del archivo. El formato del código varía según el tipo de archivo (
fileType). Para consultar un ejemplo, haga clic con el botón derecho en una tarea dentro de Operation Center y seleccione View Code.AutoRerunTimes
Integer
Sí
3
Número de veces que la tarea se vuelve a ejecutar automáticamente después de un error.
AutoRerunIntervalMillis
Integer
No
120000
Intervalo entre ejecuciones automáticas, en milisegundos. El valor máximo es 1,800,000 (30 minutos).
Este parámetro corresponde al intervalo de reejecución de la pestaña Scheduling en la DataWorks Console.
La consola muestra este valor en minutos. Asegúrese de convertir la unidad cuando use la API.
RerunMode
String
No
ALL_ALLOWED
Política de reejecución. Valores válidos:
-
ALL_ALLOWED: permite volver a ejecutar la tarea tanto si terminó correctamente como si falló.
-
FAILURE_ALLOWED: permite volver a ejecutar la tarea únicamente si falló.
-
ALL_DENIED: no permite volver a ejecutar la tarea.
Este parámetro corresponde a la propiedad de reejecución de la pestaña Scheduling en la DataWorks console.
Stop
Boolean
No
false
Indica si se pausa la programación de la tarea. Valores válidos:
-
true: pausa la programación.
-
false: no pausa la programación.
Este parámetro corresponde a la opción de pausar la programación de la pestaña Scheduling en la DataWorks Console.
ParaValue
String
No
x=a y=b z=c
Parámetros de programación.
Este parámetro corresponde a los parámetros de la pestaña Scheduling en la DataWorks Console. Para obtener información sobre cómo configurar los parámetros de programación, consulte Configurar parámetros de programación.
StartEffectDate
Long
No
936923400000
Hora de inicio de la programación automática, especificada como una marca de tiempo de Unix en milisegundos.
Este parámetro corresponde a la hora de inicio configurada en la fecha de vigencia de la pestaña Scheduling en la DataWorks console.
EndEffectDate
Long
No
4155787800000
Hora de detención de la programación automática, especificada como una marca de tiempo de Unix en milisegundos.
Este parámetro corresponde a la hora de inicio configurada en la fecha de vigencia de la pestaña Scheduling en la DataWorks console.
CronExpress
String
No
00 00-59/5 1-23 ?
Expresión cron para la programación periódica. Este parámetro corresponde a la expresión cron de la pestaña Scheduling en la DataWorks Console. DataWorks genera automáticamente una expresión cron después de configurar la programación y la hora programada.
Ejemplos:
-
Ejecutar la tarea todos los días a las 05:30:
00 30 05 ?. -
Ejecutar la tarea 15 minutos después de cada hora:
00 15 * ?. -
Ejecutar la tarea cada 10 minutos:
00 00/10 * ?. -
Ejecutar la tarea cada 10 minutos entre las 08:00 y las 17:00 todos los días:
00 00-59/10 8-17 * ?. -
Ejecutar la tarea a las 00:20 el primer día de cada mes:
00 20 00 1 * ?. -
Ejecutar la tarea cada 3 meses, a partir de las 00:10 del 1 de enero:
00 10 00 1 1-12/3 ?. -
Ejecutar la tarea a las 00:05 todos los martes y viernes:
00 05 00 2,5.
El sistema de programación de DataWorks tiene los siguientes límites para las expresiones cron:
-
El intervalo mínimo de programación es de 5 minutos.
-
La hora más temprana para programar la ejecución de una tarea es las 00:05 de cada día.
CycleType
String
No
NOT_DAY
Tipo de ciclo de programación. Los valores válidos son
NOT_DAY(minuto u hora) yDAY(día, semana o mes).Este parámetro corresponde a la programación de la pestaña Scheduling en la DataWorks Console.
DependentType
String
No
USER_DEFINE
Modo de dependencia entre ciclos. Valores válidos:
-
SELF: selecciona el nodo actual como nodo dependiente.
-
CHILD: selecciona los nodos secundarios de primer nivel como nodos dependientes.
-
USER_DEFINE: selecciona otros nodos como nodos dependientes.
-
NONE: no selecciona ningún nodo dependiente. La tarea no depende del ciclo anterior.
DependentNodeIdList
String
No
5,10,15,20
Si
DependentTypeesUSER_DEFINE, este parámetro es obligatorio. Especifique los ID de los nodos dependientes separados por comas (,).
Este parámetro corresponde a la configuración de USER_DEFINE que aparece al configurar la dependencia entre ciclos en la pestaña Scheduling de la DataWorks console.InputList
String
No
project_root,project.file1,project.001_out
Nombres de salida de los nodos anteriores de los que depende este nodo. Separe varios nombres con comas (,).
Este parámetro corresponde al nombre de salida del nodo anterior de la pestaña Scheduling en la DataWorks console.NotaEste parámetro es obligatorio al crear una tarea de sincronización por lotes con la operación
CreateDISyncTaskoUpdateFile.ProjectIdentifier
String
No
dw_project
Nombre del espacio de trabajo de DataWorks. Inicie sesión en la DataWorks Console y vaya a la página de configuración del espacio de trabajo para obtener el nombre.
Debe establecer este parámetro o el parámetroProjectIdpara especificar el espacio de trabajo de DataWorks.FileId
Long
Sí
100000001
ID del archivo. Llame a la operación
ListFilespara obtener el ID del archivo.OutputList
String
No
dw_project.ods_user_info_d
Nombre de salida del archivo.
Este parámetro corresponde al nombre de salida del nodo actual de la pestaña Scheduling en la DataWorks console.ResourceGroupIdentifier
String
No
default_group
Resource group donde se ejecuta la tarea después del despliegue. Llame a la operación
ListResourceGroupspara obtener los resource groups disponibles en el espacio de trabajo.ConnectionName
String
No
odps_source
Identificador del origen de datos que usa la tarea durante su ejecución. Llame a la operación
ListDataSourcespara obtener la lista de orígenes de datos disponibles.Owner
String
No
18023848927592
ID de usuario del propietario del archivo.
AutoParsing
Boolean
No
true
Indica si se habilita el análisis sintáctico automático del archivo. Valores válidos:
-
true: el archivo analiza el código automáticamente. -
false: el archivo no analiza el código automáticamente.
Este parámetro corresponde a la opción de analizar entradas y salidas desde el código en la pestaña Scheduling de la DataWorks Console.
SchedulerType
String
No
NORMAL
Tipo de programación. Valores válidos:
-
NORMAL: tarea programada de forma normal.
-
MANUAL: tarea manual. Se ejecuta solo cuando se activa manualmente y no se programa de forma automática.
-
PAUSE: tarea en pausa.
-
SKIP: tarea de simulación. Las tareas de simulación se programan de manera regular, pero se marcan inmediatamente como exitosas cuando inicia la programación.
AdvancedSettings
String
No
{"queue":"default","SPARK_CONF":"--conf spark.driver.memory=2g"}
Configuración avanzada de la tarea.
Este parámetro corresponde a la configuración avanzada de la página de edición de una tarea EMR Spark Streaming o EMR Streaming SQL en la DataWorks Console.
Actualmente, este parámetro solo es compatible con tareas EMR Spark Streaming y EMR Streaming SQL. El valor del parámetro debe estar en formato JSON.
StartImmediately
Boolean
No
true
Indica si la tarea se inicia inmediatamente después del despliegue. Valores válidos:
-
true: inicia la tarea inmediatamente después del despliegue.
-
false: no inicia la tarea inmediatamente después del despliegue.
Este parámetro corresponde al modo de inicio de la pestaña Scheduling, ubicada en la página de edición de una tarea EMR Spark Streaming o EMR Streaming SQL dentro de la DataWorks Console.
InputParameters
String
No
[{"ValueSource": "project_001.first_node:bizdate_param","ParameterName": "bizdate_input"}]
Parámetros de entrada de contexto del nodo. El valor es una cadena JSON. Para obtener información sobre los campos, consulte el parámetro
InputContextParameterListque devuelve la operaciónGetFile.Este parámetro corresponde a los parámetros de entrada del nodo actual en la pestaña Scheduling de la DataWorks Console.
OutputParameters
String
No
[{"Type": 1,"Value": "${bizdate}","ParameterName": "bizdate_param"}]
Parámetros de salida de contexto del nodo. El valor es una cadena JSON. Para obtener información sobre los campos, consulte el parámetro
OutputContextParameterListque devuelve la operaciónGetFile.
Este parámetro corresponde a los parámetros de salida del nodo actual en la pestaña Scheduling de la DataWorks Console. -
-
Envíe la tarea de Data Integration.
Llame a la operación
SubmitFilepara enviar la tarea de Data Integration al entorno de desarrollo del sistema de programación. La respuesta devuelve undeploymentId. Utilice este ID con la operaciónGetDeploymentpara recuperar los detalles del despliegue.public void submitFile() throws ClientException{ SubmitFileRequest request = new SubmitFileRequest(); request.setProjectId(78837L); request.setProjectIdentifier("zxy_8221431"); // This node ID is the ID that is returned when you create the node. It corresponds to the file_id in the File table of the database. request.setFileId(501576542L); request.setComment("Comment"); SubmitFileResponse acsResponse = client.getAcsResponse(request); // The DeploymentId is the return value of the commit or publish operation. Long deploymentId = acsResponse.getData(); log.info(acsResponse.toString()); }El código anterior muestra un ejemplo de cómo configurar algunos de los parámetros. Para obtener más información sobre los parámetros, consulte SubmitFile y GetDeployment.
-
Despliegue la tarea de sincronización en el entorno de producción.
Llame a la operación
DeployFilepara publicar la tarea de sincronización de Data Integration en el entorno de producción.NotaEsta operación solo aplica a espacios de trabajo en modo estándar.
public void deploy() throws ClientException{ DeployFileRequest request = new DeployFileRequest(); request.setProjectIdentifier("zxy_8221431"); request.setFileId(501576542L); request.setComment("Comment"); // Specify either NodeId or file_id. The value of NodeId is the node ID in the basic properties of the scheduling configuration. request.setNodeId(700004537241L); DeployFileResponse acsResponse = client.getAcsResponse(request); // The DeploymentId is the return value of the commit or publish operation. Long deploymentId = acsResponse.getData(); log.info(acsResponse.getData().toString()); }El código anterior muestra un ejemplo de cómo configurar algunos de los parámetros. Para obtener más información sobre los parámetros, consulte DeployFile.
-
Obtenga los detalles del paquete de despliegue.
La operación devuelve un
deploymentId. Utilice este ID con la operaciónGetDeploymentpara comprobar el estado del despliegue. UnStatuscon valor1indica que el despliegue fue exitoso.public void getDeployment() throws ClientException{ GetDeploymentRequest request = new GetDeploymentRequest(); request.setProjectId(78837L); request.setProjectIdentifier("zxy_8221431"); // The DeploymentId is the return value of the commit or publish operation. Call the GetDeployment operation to get the details of this deployment. request.setDeploymentId(2776067L); GetDeploymentResponse acsResponse = client.getAcsResponse(request); log.info(acsResponse.getData().toString()); }El código anterior muestra un ejemplo de cómo configurar algunos de los parámetros. Para obtener más información sobre los parámetros, consulte GetDeployment.
Modificar la configuración de una tarea de sincronización
Para modificar una tarea, llame a la operación UpdateDISyncTask y actualice el Content de la tarea o bien actualice el resource group exclusivo mediante el parámetro TaskParam. Después de actualizar la tarea, debe enviarla y desplegarla nuevamente. Para obtener más información, consulte Procedure.
Eliminar una tarea de sincronización
Para eliminar una tarea de sincronización de Data Integration, llame a la operación DeleteFile.
Código de ejemplo
import com.alibaba.fastjson.JSONObject;
import com.aliyun.dataworks_public20200518.Client;
import com.aliyun.dataworks_public20200518.models.*;
import com.aliyun.teaopenapi.models.Config;
public class createofflineTask {
static Long createTask(String fileName) throws Exception {
Long projectId = 2043L;
String taskType = "DI_OFFLINE";
String taskContent = "{\n" +
" \"type\": \"job\",\n" +
" \"version\": \"2.0\",\n" +
" \"steps\": [\n" +
" {\n" +
" \"stepType\": \"mysql\",\n" +
" \"parameter\": {\n" +
" \"envType\": 0,\n" +
" \"datasource\": \"mysql_autotest_dev\",\n" +
" \"column\": [\n" +
" \"id\",\n" +
" \"name\"\n" +
" ],\n" +
" \"connection\": [\n" +
" {\n" +
" \"datasource\": \"mysql_autotest_dev\",\n" +
" \"table\": [\n" +
" \"user\"\n" +
" ]\n" +
" }\n" +
" ],\n" +
" \"where\": \"\",\n" +
" \"splitPk\": \"\",\n" +
" \"encoding\": \"UTF-8\"\n" +
" },\n" +
" \"name\": \"Reader\",\n" +
" \"category\": \"reader\"\n" +
" },\n" +
" {\n" +
" \"stepType\": \"odps\",\n" +
" \"parameter\": {\n" +
" \"partition\": \"pt=${bizdate}\",\n" +
" \"truncate\": true,\n" +
" \"datasource\": \"odps_source\",\n" +
" \"envType\": 0,\n" +
" \"column\": [\n" +
" \"id\",\n" +
" \"name\"\n" +
" ],\n" +
" \"emptyAsNull\": false,\n" +
" \"tableComment\": \"null\",\n" +
" \"table\": \"user\"\n" +
" },\n" +
" \"name\": \"Writer\",\n" +
" \"category\": \"writer\"\n" +
" }\n" +
" ],\n" +
" \"setting\": {\n" +
" \"executeMode\": null,\n" +
" \"errorLimit\": {\n" +
" \"record\": \"\"\n" +
" },\n" +
" \"speed\": {\n" +
" \"concurrent\": 2,\n" +
" \"throttle\": false\n" +
" }\n" +
" },\n" +
" \"order\": {\n" +
" \"hops\": [\n" +
" {\n" +
" \"from\": \"Reader\",\n" +
" \"to\": \"Writer\"\n" +
" }\n" +
" ]\n" +
" }\n" +
"}";
CreateDISyncTaskRequest request = new CreateDISyncTaskRequest();
request.setProjectId(projectId);
request.setTaskType(taskType);
request.setTaskContent(taskContent);
request.setTaskName(fileName);
request.setTaskParam("{\"FileFolderPath\":\"Workflow/Automated_Testing_Workspace_Do_Not_Modify/Data_Integration\",\"ResourceGroup\":\"S_res_group_XXX\"}");
// Use an exclusive resource group for Data Integration.
CreateDISyncTaskResponse response1 = client.getAcsResponse(request);
return response1.getData().getFileId();
}
public static void updateFile(Long fileId) throws Exception {
UpdateFileRequest request = new UpdateFileRequest();
request.setProjectId(2043L);
request.setFileId(fileId);
request.setAutoRerunTimes(3);
request.setRerunMode("FAILURE_ALLOWED");
request.setCronExpress("00 30 05 * * ?");
request.setCycleType("DAY");
request.setResourceGroupIdentifier("S_res_group_XXX");
// Use an exclusive resource group for scheduling.
request.setInputList("dataworks_di_autotest_root");
request.setAutoParsing(true);
request.setDependentNodeIdList("5,10,15,20");
request.setDependentType("SELF");
request.setStartEffectDate(0L);
request.setEndEffectDate(4155787800000L);
request.setFileDescription("description");
request.setStop(false);
request.setParaValue("x=a y=b z=c");
request.setSchedulerType("NORMAL");
request.setAutoRerunIntervalMillis(120000);
UpdateFileResponse response1 = client.getAcsResponse(request);
}
static void deleteTask(Long fileId) throws Exception {
DeleteFileRequest request = new DeleteFileRequest();
Long projectId = 63845L;
request.setProjectId(projectId);
request.setFileId(fileId);
String akId = "XXX";
String akSecret = "XXXX";
String regionId = "cn-hangzhou";
IClientProfile profile = DefaultProfile.getProfile(regionId, akId, akSecret);
DefaultProfile.addEndpoint("cn-hangzhou","dataworks-public","dataworks.cn-hangzhou.aliyuncs.com");
IAcsClient client;
client = new DefaultAcsClient(profile);
DeleteFileResponse response1 = client.getAcsResponse(request);
System.out.println(JSONObject.toJSONString(response1));
}
static IAcsClient client;
public static void main(String[] args) throws Exception {
String akId = "XX";
// Please ensure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.https://www.alibabacloud.com/help/zh/alibaba-cloud-sdk-262060/latest/configure-credentials-378659
IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
DefaultProfile.addEndpoint(regionId, "dataworks-public", "dataworks." + regionId + ".aliyuncs.com");
client = new DefaultAcsClient(profile);
String taskName = "offline_job_0930_1648";
Long fileId = createTask(taskName); // Create a data integration task.
updateFile(fileId); // Modify the scheduling properties of the data integration task.
}
}