关于通过java调用datax,返回任务执行的方法

要通过Java调用DataX并返回任务执行的方法，以下是完整的攻略：

1. 导入DataX依赖

需要在Java项目中添加DataX的依赖，可以通过Maven或者Gradle实现：

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>datax</artifactId>
    <version>${datax.version}</version>
</dependency>

其中 datax.version 需要自行根据实际情况设置。

1. 编写DataX配置文件

需要编写DataX的配置文件，指定数据源、数据目的地、数据同步任务等。

这里以从MySQL中读取数据并插入到Elasticsearch中为例，配置文件如下：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "123456",
            "column": ["id", "name", "age"],
            "connection": [
              {
                "jdbcUrl": "jdbc:mysql://localhost:3306/test",
                "table": ["user"]
              }
            ]
          }
        },
        "writer": {
          "name": "eswriter",
          "parameter": {
            "endpoint": "http://localhost:9200",
            "index": "user",
            "type": "doc",
            "column": ["id", "name", "age"]
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "throttle": true,
        "concurrent": 2
      }
    }
  }
}

需要将配置文件保存为 json 文件，例如 job.json。

1. 编写Java代码调用DataX

可以使用DataX自带的 Engine 类来启动DataX任务。需要先将配置文件转换为 JSONObject 对象，然后传入 Engine 的 transform 方法中，代码示例如下：

import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.Engine;

import java.util.Properties;

public class DataXDemo {

    public static void main(String[] args) {
        Properties properties = System.getProperties();
        properties.setProperty("datax.home", "/path/to/datax");
        System.setProperties(properties);

        try {
            Configuration configuration = Configuration.from("{\n" +
                    "  \"job\": {\n" +
                    "    \"content\": [\n" +
                    "      {\n" +
                    "        \"reader\": {\n" +
                    "          \"name\": \"mysqlreader\",\n" +
                    "          \"parameter\": {\n" +
                    "            \"username\": \"root\",\n" +
                    "            \"password\": \"123456\",\n" +
                    "            \"column\": [\"id\", \"name\", \"age\"],\n" +
                    "            \"connection\": [\n" +
                    "              {\n" +
                    "                \"jdbcUrl\": \"jdbc:mysql://localhost:3306/test\",\n" +
                    "                \"table\": [\"user\"]\n" +
                    "              }\n" +
                    "            ]\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"writer\": {\n" +
                    "          \"name\": \"eswriter\",\n" +
                    "          \"parameter\": {\n" +
                    "            \"endpoint\": \"http://localhost:9200\",\n" +
                    "            \"index\": \"user\",\n" +
                    "            \"type\": \"doc\",\n" +
                    "            \"column\": [\"id\", \"name\", \"age\"]\n" +
                    "          }\n" +
                    "        }\n" +
                    "      }\n" +
                    "    ],\n" +
                    "    \"setting\": {\n" +
                    "      \"speed\": {\n" +
                    "        \"throttle\": true,\n" +
                    "        \"concurrent\": 2\n" +
                    "      }\n" +
                    "    }\n" +
                    "  }\n" +
                    "}");

            Engine engine = new Engine();
            engine.transform(configuration);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

需要将其中的 /path/to/datax 修改为你本地DataX的安装路径，然后运行代码即可启动DataX任务。在任务执行完成后，会在日志中打印出执行结果等信息。

另外，还可以将任务执行结果以JSON的形式返回，代码示例如下：

import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.Engine;
import com.alibaba.datax.core.job.JobContainer;
import com.alibaba.datax.core.job.jobinfo.JobInfo;
import com.alibaba.datax.core.util.container.CoreConstant;

import java.util.Properties;

public class DataXDemo {

    public static void main(String[] args) {
        Properties properties = System.getProperties();
        properties.setProperty(CoreConstant.DATAX_HOME, "/path/to/datax");
        System.setProperties(properties);

        try {
            Configuration configuration = Configuration.from("{\n" +
                    "  \"job\": {\n" +
                    "    \"content\": [\n" +
                    "      {\n" +
                    "        \"reader\": {\n" +
                    "          \"name\": \"mysqlreader\",\n" +
                    "          \"parameter\": {\n" +
                    "            \"username\": \"root\",\n" +
                    "            \"password\": \"123456\",\n" +
                    "            \"column\": [\"id\", \"name\", \"age\"],\n" +
                    "            \"connection\": [\n" +
                    "              {\n" +
                    "                \"jdbcUrl\": \"jdbc:mysql://localhost:3306/test\",\n" +
                    "                \"table\": [\"user\"]\n" +
                    "              }\n" +
                    "            ]\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"writer\": {\n" +
                    "          \"name\": \"eswriter\",\n" +
                    "          \"parameter\": {\n" +
                    "            \"endpoint\": \"http://localhost:9200\",\n" +
                    "            \"index\": \"user\",\n" +
                    "            \"type\": \"doc\",\n" +
                    "            \"column\": [\"id\", \"name\", \"age\"]\n" +
                    "          }\n" +
                    "        }\n" +
                    "      }\n" +
                    "    ],\n" +
                    "    \"setting\": {\n" +
                    "      \"speed\": {\n" +
                    "        \"throttle\": true,\n" +
                    "        \"concurrent\": 2\n" +
                    "      }\n" +
                    "    }\n" +
                    "  }\n" +
                    "}");

            Engine engine = new Engine();
            JobContainer jobContainer = engine.createJobContainer(configuration);
            JobInfo jobInfo = jobContainer.getJobInfo();
            jobContainer.start();

            while (jobContainer.getState() != JobContainer.State.SUCCEEDED) {
                if (jobContainer.getState() == JobContainer.State.FAILED) {
                    throw new RuntimeException("DataX task failed");
                }
                Thread.sleep(1000);
            }

            Object result = jobInfo.getTaskResult();

            System.out.println(result);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

在上述代码中，通过调用 JobContainer 的 getTaskResult() 方法可以获取执行结果，将结果以JSON的格式输出。