java - 在 Java 中使用 Sqoop 将 MYSQL 表导入 hdfs
问题描述
我正在尝试使用 JAVA 中的勺子将表从 MYSQL 导入 HDFS。
我在本地机器上安装了勺子。
我的系统中有hdfs。
我在mysql中创建了一个数据库和表。我在 pom.xml 中添加了 sqoop 依赖项。
我试着做这个代码:
//Here I am using a table Persons, with columns PersonID and LastName
import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MConnection;
import org.apache.sqoop.model.MConnectionForms;
import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MJobForms;
import org.apache.sqoop.model.MSubmission;
import org.apache.sqoop.validation.Status;
/**
* @author devan
* @date 19-Sep-2013
* @mail msdevanms@gmail.com
*/
public class SqoopImport {
public static void main(String[] args) {
String connectionString = "jdbc:mysql://YourMysqlIP:3306/test";
String username = "YourMysqUserName";
String password = "YourMysqlPassword";
String schemaName = "YourMysqlDB";
String tableName = "Persons";
String columns = "PersonID,LastName"; //comma seperated column names
String partitionColumn = "PersonID";
String outputDirectory = "/output/Persons";
String url = "http://YourSqoopIP:12000/sqoop/";
SqoopClient client = new SqoopClient(url);
//client.setServerUrl(newUrl);
//Dummy connection object
MConnection newCon = client.newConnection(1);
//Get connection and framework forms. Set name for connection
MConnectionForms conForms = newCon.getConnectorPart();
MConnectionForms frameworkForms = newCon.getFrameworkPart();
newCon.setName("MyConnection");
//Set connection forms values
conForms.getStringInput("connection.connectionString").setValue(connectionString);
conForms.getStringInput("connection.jdbcDriver").setValue("com.mysql.jdbc.Driver");
conForms.getStringInput("connection.username").setValue(username);
conForms.getStringInput("connection.password").setValue(password);
//frameworkForms.getIntegerInput("security.maxConnections").setValue(0);
Status status = client.createConnection(newCon);
if (status.canProceed()) {
System.out.println("Created. New Connection ID : " + newCon.getPersistenceId());
} else {
System.out.println("Check for status and forms error ");
}
//Creating dummy job object
MJob newjob = client.newJob(newCon.getPersistenceId(), org.apache.sqoop.model.MJob.Type.IMPORT);
MJobForms connectorForm = newjob.getConnectorPart();
MJobForms frameworkForm = newjob.getFrameworkPart();
newjob.setName("ImportJob");
//Database configuration
connectorForm.getStringInput("table.schemaName").setValue(schemaName);
//Input either table name or sql
connectorForm.getStringInput("table.tableName").setValue(tableName);
//connectorForm.getStringInput("table.sql").setValue("select id,name from table where ${CONDITIONS}");
connectorForm.getStringInput("table.columns").setValue(columns);
connectorForm.getStringInput("table.partitionColumn").setValue(partitionColumn);
//Set boundary value only if required
//connectorForm.getStringInput("table.boundaryQuery").setValue("");
//Output configurations
frameworkForm.getEnumInput("output.storageType").setValue("HDFS");
frameworkForm.getEnumInput("output.outputFormat").setValue("TEXT_FILE");//Other option: SEQUENCE_FILE / TEXT_FILE
frameworkForm.getStringInput("output.outputDirectory").setValue(outputDirectory);
//Job resources
frameworkForm.getIntegerInput("throttling.extractors").setValue(1);
frameworkForm.getIntegerInput("throttling.loaders").setValue(1);
status = client.createJob(newjob);
if (status.canProceed()) {
System.out.println("New Job ID: " + newjob.getPersistenceId());
} else {
System.out.println("Check for status and forms error ");
}
//Now Submit the Job
MSubmission submission = client.startSubmission(newjob.getPersistenceId());
System.out.println("Status : " + submission.getStatus());
}
}
任何帮助,将不胜感激。
解决方案
推荐阅读
- php - 在 Moodle 中添加匹配测验时,我在选项中遇到了这个问题
- python - Python字典转换
- python - 根据条件 python pandas 创建列
- jupyter-notebook - 有没有办法通过使用 Mathjax 的外部包在 jupyter notebook 中写下化学方程式?
- amazon-s3 - 将多个文件夹上传到 Amazon S3 存储桶
- java - 信息:检测到的方言:W3C [Utils] [ERROR] [Error] java.lang.NullPointerException
- node.js - Promises and async await is not working -Unhandled Rejection at: Promise Promise {
'noData' } reason: undefined - node.js - mysql - mysql使用phpmyadmin在存储过程中调用函数
- c# - 如何在 C# 中就耦合重构相同的返回类型?
- asp.net-core - 为什么剃须刀组件中的HttpContext为空?