Sqoop 引擎

本文主要介绍Sqoop(>=1.1.2 版本支持)引擎的配置、部署和使用。

1 Sqoop引擎Linkis系统参数配置

Sqoop引擎主要依赖Hadoop基础环境，如果该节点需要部署Sqoop引擎，需要部署Hadoop客户端环境。

强烈建议您在执行Sqoop任务之前，先在该节点使用原生的Sqoop执行测试任务，以检测该节点环境是否正常。

1. #验证sqoop环境是否可用 参考示例：将hdfs的/user/hive/warehouse/hadoop/test_linkis_sqoop文件数据导入到mysql表 test_sqoop中

4. sqoop export \
5. --connect  jdbc:mysql://10.10.10.10/test \
6. --username test \
7. --password test123\
8. --table test_sqoop \
9. --columns user_id,user_code,user_name,email,status \
10. --export-dir /user/hive/warehouse/hadoop/test_linkis_sqoop  \
11. --update-mode allowinsert  \
12. --verbose ;

表1-1 环境配置清单

2 Sqoop引擎的配置和部署

2.1 Sqoop版本的选择和编译

Linkis 1.1.2及以上支持的主流Sqoop版本1.4.6与1.4.7，更高版本可能需要修改部分代码重新编译。

2.2 Sqoop engineConn部署和加载

注意: 编译sqoop引擎之前需要进行linkis项目全量编译

1. 单独编译sqoop的方式
2. ${linkis_code_dir}/linkis-engineconn-plugins/engineconn-plugins/sqoop/
3. mvn clean install

安装方式是将编译出来的引擎包,位置在

1. ${linkis_code_dir}/linkis-enginepconn-plugins/engineconn-plugins/sqoop/target/sqoop-engineconn.zip

然后上传部署到linkis服务器

1. ${LINKIS_HOME}/lib/linkis-engineplugins

并重启linkis-engineplugin

1. cd ${LINKIS_HOME}/sbin
2. sh linkis-daemon.sh restart cg-engineplugin

engineplugin更详细的介绍可以参看下面的文章。

<https://linkis.apache.org/zh-CN/docs/latest/deployment/engine-conn-plugin-installation>

3 Sqoop引擎的使用

3.1 通过Linkis-cli进行任务提交

hdfs文件导出到mysql

1. sh linkis-cli-sqoop export \
2. -D mapreduce.job.queuename=ide \
3. --connect jdbc:mysql://10.10.10.10:9600/testdb \
4. --username password@123 \
5. --password password@123  \
6. --table test_sqoop_01_copy \
7. --columns user_id,user_code,user_name,email,status \
8. --export-dir /user/hive/warehouse/hadoop/test_linkis_sqoop_2 \
9. --update-mode allowinsert --verbose ;

mysql数据导入到hive库

1. mysql导入到hive 库linkis_test_ind.test_import_sqoop_1,表test_import_sqoop_1不存在 需要添加参数 --create-hive-table

4. sh linkis-cli-sqoop import -D mapreduce.job.queuename=dws \
5. --connect jdbc:mysql://10.10.10.10:3306/casion_test \
6. --username hadoop \
7. --password password@123 \
8. --table test_sqoop_01 \
9. --columns user_id,user_code,user_name,email,status \
10. --fields-terminated-by ',' \
11. --hive-import --create-hive-table \
12. --hive-database casionxia_ind \
13. --hive-table test_import_sqoop_1 \
14. --hive-drop-import-delims \
15. --delete-target-dir \
16. --input-null-non-string '\\N' \
17. --input-null-string '\\N' \
18. --verbose ;

23. mysql导入到hive 库linkis_test_ind.test_import_sqoop_1,表test_import_sqoop_1存在 移除参数--create-hive-table \
24. sh linkis-cli-sqoop import -D mapreduce.job.queuename=dws \
25. --connect jdbc:mysql://10.10.10.10:9600/testdb \
26. --username testdb \
27. --password password@123 \
28. --table test_sqoop_01 \
29. --columns user_id,user_code,user_name,email,status \
30. --fields-terminated-by ',' \
31. --hive-import \
32. --hive-database linkis_test_ind \
33. --hive-table test_import_sqoop_1 \
34. --hive-overwrite \
35. --hive-drop-import-delims \
36. --delete-target-dir \
37. --input-null-non-string '\\N' \
38. --input-null-string '\\N' \
39. --verbose ;

3.1 OnceEngineConn方式

OnceEngineConn的使用方式是通过LinkisManagerClient调用LinkisManager的createEngineConn的接口，并将代码发给创建的Sqoop引擎，然后Sqoop引擎就开始执行，此方式可以被其他系统进行调用，比如Exchangis。Client的使用方式也很简单，首先新建一个maven项目，或者在您的项目中引入以下的依赖

1. <dependency>
2. <groupId>org.apache.linkis</groupId>
3. <artifactId>linkis-computation-client</artifactId>
4. <version>${linkis.version}</version>
5. </dependency>

测试用例：

3. import java.util.concurrent.TimeUnit

6. import java.util

9. import org.apache.linkis.computation.client.LinkisJobBuilder
10. import org.apache.linkis.computation.client.once.simple.{SimpleOnceJob, SimpleOnceJobBuilder, SubmittableSimpleOnceJob}
11. import org.apache.linkis.computation.client.operator.impl.{EngineConnLogOperator, EngineConnMetricsOperator, EngineConnProgressOperator}
12. import org.apache.linkis.computation.client.utils.LabelKeyUtils

15. import scala.collection.JavaConverters._

18. object SqoopOnceJobTest extends App {
19. LinkisJobBuilder.setDefaultServerUrl("http://127.0.0.1:9001")
20. val logPath = "C:\\Users\\resources\\log4j.properties"
21. System.setProperty("log4j.configurationFile", logPath)
22. val startUpMap = new util.HashMap[String, Any]
23. startUpMap.put("wds.linkis.engineconn.java.driver.memory", "1g")
24. val builder = SimpleOnceJob.builder().setCreateService("Linkis-Client")
25. .addLabel(LabelKeyUtils.ENGINE_TYPE_LABEL_KEY, "sqoop-1.4.6")
26. .addLabel(LabelKeyUtils.USER_CREATOR_LABEL_KEY, "Client")
27. .addLabel(LabelKeyUtils.ENGINE_CONN_MODE_LABEL_KEY, "once")
28. .setStartupParams(startUpMap)
29. .setMaxSubmitTime(30000)
30. .addExecuteUser("freeuser")
31. val onceJob = importJob(builder)
32. val time = System.currentTimeMillis()
33. onceJob.submit()

36. println(onceJob.getId)
37. val logOperator = onceJob.getOperator(EngineConnLogOperator.OPERATOR_NAME).asInstanceOf[EngineConnLogOperator]
38. println(onceJob.getECMServiceInstance)
39. logOperator.setFromLine(0)
40. logOperator.setECMServiceInstance(onceJob.getECMServiceInstance)
41. logOperator.setEngineConnType("sqoop")
42. logOperator.setIgnoreKeywords("[main],[SpringContextShutdownHook]")
43. var progressOperator = onceJob.getOperator(EngineConnProgressOperator.OPERATOR_NAME).asInstanceOf[EngineConnProgressOperator]
44. var metricOperator = onceJob.getOperator(EngineConnMetricsOperator.OPERATOR_NAME).asInstanceOf[EngineConnMetricsOperator]
45. var end = false
46. var rowBefore = 1
47. while (!end || rowBefore > 0){
48. if(onceJob.isCompleted) {
49. end = true
50. metricOperator = null
51. }
52. logOperator.setPageSize(100)
53. Utils.tryQuietly{
54. val logs = logOperator.apply()
55. logs.logs.asScala.foreach( log => {
56. println(log)
57. })
58. rowBefore = logs.logs.size
59. }
60. Thread.sleep(3000)
61. Option(metricOperator).foreach( operator => {
62. if (!onceJob.isCompleted){
63. println(s"Metric监控: ${operator.apply()}")
64. println(s"进度: ${progressOperator.apply()}")
65. }
66. })
67. }
68. onceJob.isCompleted
69. onceJob.waitForCompleted()
70. println(onceJob.getStatus)
71. println(TimeUnit.SECONDS.convert(System.currentTimeMillis() - time, TimeUnit.MILLISECONDS) + "s")
72. System.exit(0)

77. def importJob(jobBuilder: SimpleOnceJobBuilder): SubmittableSimpleOnceJob = {
78. jobBuilder
79. .addJobContent("sqoop.env.mapreduce.job.queuename", "queue_10")
80. .addJobContent("sqoop.mode", "import")
81. .addJobContent("sqoop.args.connect", "jdbc:mysql://127.0.0.1:3306/exchangis")
82. .addJobContent("sqoop.args.username", "free")
83. .addJobContent("sqoop.args.password", "testpwd")
84. .addJobContent("sqoop.args.query", "select id as order_number, sno as time from" +
85. " exchangis where sno =1 and $CONDITIONS")
86. .addJobContent("sqoop.args.hcatalog.database", "freedb")
87. .addJobContent("sqoop.args.hcatalog.table", "zy_test")
88. .addJobContent("sqoop.args.hcatalog.partition.keys", "month")
89. .addJobContent("sqoop.args.hcatalog.partition.values", "3")
90. .addJobContent("sqoop.args.num.mappers", "1")
91. .build()
92. }

95. def exportJob(jobBuilder: SimpleOnceJobBuilder): SubmittableSimpleOnceJob = {
96. jobBuilder
97. .addJobContent("sqoop.env.mapreduce.job.queuename", "queue1")
98. .addJobContent("sqoop.mode", "import")
99. .addJobContent("sqoop.args.connect", "jdbc:mysql://127.0.0.1:3306/exchangis")
100. .addJobContent("sqoop.args.query", "select id as order, sno as great_time from" +
101. " exchangis_table where sno =1 and $CONDITIONS")
102. .addJobContent("sqoop.args.hcatalog.database", "hadoop")
103. .addJobContent("sqoop.args.hcatalog.table", "partition_33")
104. .addJobContent("sqoop.args.hcatalog.partition.keys", "month")
105. .addJobContent("sqoop.args.hcatalog.partition.values", "4")
106. .addJobContent("sqoop.args.num.mappers", "1")
107. .build()
108. }

4 参数

4.1 Common arguments

4.2 Export import arguments

4.3 Import control arguments

4.4 Incremental import argument

4.5 Output line formatting arguments

4.6 Input parsing arguments

4.7 Hive arguments

4.8 HBase arguments

4.9 HCatalog arguments

4.10 Accumulo arguments

4.11 Code generation arguments

4.12 Generic Hadoop command-line arguments

must preceed any tool-specific arguments,Generic options supported are