当前位置：首页 > Java

java如何调用sqoop

2026-03-22 13:23:48Java

Java 调用 Sqoop 的方法

Java 可以通过命令行调用或直接使用 Sqoop API 来执行 Sqoop 操作。以下是两种常见的方法：

使用 Runtime 执行命令行调用

通过 Java 的 Runtime 类或 ProcessBuilder 执行 Sqoop 命令行指令，适合简单的 Sqoop 任务。

import java.io.BufferedReader;
import java.io.InputStreamReader;

public class SqoopJavaExample {
    public static void main(String[] args) {
        try {
            // 构建 Sqoop 导入命令
            String[] command = {
                "sqoop",
                "import",
                "--connect", "jdbc:mysql://localhost/mydatabase",
                "--username", "user",
                "--password", "password",
                "--table", "employees",
                "--target-dir", "/user/hadoop/employees"
            };

            // 执行命令
            ProcessBuilder processBuilder = new ProcessBuilder(command);
            Process process = processBuilder.start();

            // 读取输出
            BufferedReader reader = new BufferedReader(
                new InputStreamReader(process.getInputStream())
            );
            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println(line);
            }

            // 等待命令执行完成
            int exitCode = process.waitFor();
            System.out.println("Exited with code: " + exitCode);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

使用 Sqoop API

Sqoop 提供了 Java API，可以直接在代码中配置和运行 Sqoop 任务，适合复杂场景。

添加依赖
在 Maven 项目中添加 Sqoop 依赖：

<dependency>
    <groupId>org.apache.sqoop</groupId>
    <artifactId>sqoop-client</artifactId>
    <version>1.99.7</version>
</dependency>

代码示例
使用 SqoopClient 配置并提交任务：

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MLink;
import org.apache.sqoop.model.MConfig;
import org.apache.sqoop.validation.Status;

public class SqoopAPIClient {
    public static void main(String[] args) {
        String serverUrl = "http://localhost:12000/sqoop/";
        SqoopClient client = new SqoopClient(serverUrl);

        // 创建源数据库链接配置
        MLink jdbcLink = client.createLink("jdbc-connector");
        jdbcLink.getConnectorLinkConfig().getStringInput("linkConfig.jdbcDriver").setValue("com.mysql.jdbc.Driver");
        jdbcLink.getConnectorLinkConfig().getStringInput("linkConfig.connectionString").setValue("jdbc:mysql://localhost/mydatabase");
        jdbcLink.getConnectorLinkConfig().getStringInput("linkConfig.username").setValue("user");
        jdbcLink.getConnectorLinkConfig().getStringInput("linkConfig.password").setValue("password");
        Status jdbcStatus = client.saveLink(jdbcLink);

        // 创建 HDFS 目标链接配置
        MLink hdfsLink = client.createLink("hdfs-connector");
        hdfsLink.getConnectorLinkConfig().getStringInput("linkConfig.uri").setValue("hdfs://localhost:8020");
        Status hdfsStatus = client.saveLink(hdfsLink);

        // 创建导入任务
        MJob job = client.createJob(jdbcLink.getPersistenceId(), hdfsLink.getPersistenceId());
        job.getFromJobConfig().getStringInput("fromJobConfig.schemaName").setValue("mydatabase");
        job.getFromJobConfig().getStringInput("fromJobConfig.tableName").setValue("employees");
        job.getToJobConfig().getStringInput("toJobConfig.outputDirectory").setValue("/user/hadoop/employees");
        Status jobStatus = client.saveJob(job);

        // 提交任务
        client.startJob(job.getPersistenceId());
    }
}