Sqoop 引擎

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

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

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

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

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

表1-1 环境配置清单

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

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

1. 单独编译sqoop的方式
2. ${linkis_code_dir}/linkis-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/install-engineconn

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

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

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

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>

测试用例：

2. import java.util.concurrent.TimeUnit

4. import java.util

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

11. import scala.collection.JavaConverters._

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

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

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

86. def exportJob(jobBuilder: SimpleOnceJobBuilder): SubmittableSimpleOnceJob = {
87. jobBuilder
88. .addJobContent("sqoop.env.mapreduce.job.queuename", "queue1")
89. .addJobContent("sqoop.mode", "import")
90. .addJobContent("sqoop.args.connect", "jdbc:mysql://127.0.0.1:3306/exchangis")
91. .addJobContent("sqoop.args.query", "select id as order, sno as great_time from" +
92. " exchangis_table where sno =1 and $CONDITIONS")
93. .addJobContent("sqoop.args.hcatalog.database", "hadoop")
94. .addJobContent("sqoop.args.hcatalog.table", "partition_33")
95. .addJobContent("sqoop.args.hcatalog.partition.keys", "month")
96. .addJobContent("sqoop.args.hcatalog.partition.values", "4")
97. .addJobContent("sqoop.args.num.mappers", "1")
98. .build()
99. }

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