Sqoop Engine usage documentation

This article mainly introduces the configuration, deployment and use of the Sqoop engine in Linkis1.X.

1.Sqoop engine Linkis system parameter configuration

The Sqoop engine mainly depends on the Hadoop basic environment. If the node needs to deploy the Sqoop engine, the Hadoop client environment needs to be deployed.

It is strongly recommended that you use the native Sqoop to execute the test task on the node before executing the Sqoop task to check whether the node environment is normal.

表1-1 环境配置清单

2.Sqoop Engine configuration and deployment

2.1 Sqoop Version selection and compilation

Mainstream Sqoop versions 1.4.6 and 1.4.7 supported by Linkis 1.1.2 and above, and later versions may need to modify some code and recompile

2.2 Sqoop engineConn deploy and load

Note: Before compiling the sqoop engine, the linkis project needs to be fully compiled

1. Compile sqoop separately:
2. ${linkis_code_dir}linkis-engineconn-plugins/sqoop/
3. mvn clean install

The installation method is to compile the compiled engine package, located in

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

and then deploy to

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

and restart linkis-engineplugin

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

More engineplugin details can be found in the following article. <https://linkis.apache.org/zh-CN/docs/1.1.1/deployment/engine-conn-plugin-installation>

3.Sqoop Engine Usage

3.1 OnceEngineConn

OnceEngineConn is used by calling LinkisManager’s createEngineConn interface through LinkisManagerClient, and sending the code to the created Sqoop engine, and then the Sqoop engine starts to execute. This method can be called by other systems, such as Exchange. The use of Client is also very simple, first create a new maven project, or introduce the following dependencies into your project

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

Test Case：

3. package com.webank.wedatasphere.exchangis.job.server.log.client

6. import java.util.concurrent.TimeUnit

9. import java.util

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

18. import scala.collection.JavaConverters._

21. object SqoopOnceJobTest extends App {
22. LinkisJobBuilder.setDefaultServerUrl("http://127.0.0.1:9001")
23. val logPath = "C:\\Users\\resources\\log4j.properties"
24. System.setProperty("log4j.configurationFile", logPath)
25. val startUpMap = new util.HashMap[String, Any]
26. startUpMap.put("wds.linkis.engineconn.java.driver.memory", "1g")
27. val builder = SimpleOnceJob.builder().setCreateService("Linkis-Client")
28. .addLabel(LabelKeyUtils.ENGINE_TYPE_LABEL_KEY, "sqoop-1.4.6")
29. .addLabel(LabelKeyUtils.USER_CREATOR_LABEL_KEY, "Client")
30. .addLabel(LabelKeyUtils.ENGINE_CONN_MODE_LABEL_KEY, "once")
31. .setStartupParams(startUpMap)
32. .setMaxSubmitTime(30000)
33. .addExecuteUser("freeuser")
34. val onceJob = importJob(builder)
35. val time = System.currentTimeMillis()
36. onceJob.submit()
37. println(onceJob.getId)
38. val logOperator = onceJob.getOperator(EngineConnLogOperator.OPERATOR_NAME).asInstanceOf[EngineConnLogOperator]
39. println(onceJob.getECMServiceInstance)
40. logOperator.setFromLine(0)
41. logOperator.setECMServiceInstance(onceJob.getECMServiceInstance)
42. logOperator.setEngineConnType("sqoop")
43. logOperator.setIgnoreKeywords("[main],[SpringContextShutdownHook]")
44. var progressOperator = onceJob.getOperator(EngineConnProgressOperator.OPERATOR_NAME).asInstanceOf[EngineConnProgressOperator]
45. var metricOperator = onceJob.getOperator(EngineConnMetricsOperator.OPERATOR_NAME).asInstanceOf[EngineConnMetricsOperator]
46. var end = false
47. var rowBefore = 1
48. while (!end || rowBefore > 0){
49. if(onceJob.isCompleted) {
50. end = true
51. metricOperator = null
52. }
53. logOperator.setPageSize(100)
54. Utils.tryQuietly{
55. val logs = logOperator.apply()
56. logs.logs.asScala.foreach( log => {
57. println(log)
58. })
59. rowBefore = logs.logs.size
60. }
61. Thread.sleep(3000)
62. Option(metricOperator).foreach( operator => {
63. if (!onceJob.isCompleted){
64. println(s"Metric Monitor: ${operator.apply()}")
65. println(s"Progress: ${progressOperator.apply()}")
66. }
67. })
68. }
69. onceJob.isCompleted
70. onceJob.waitForCompleted()
71. println(onceJob.getStatus)
72. println(TimeUnit.SECONDS.convert(System.currentTimeMillis() - time, TimeUnit.MILLISECONDS) + "s")
73. System.exit(0)

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

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

Parameter Comparison table (with native parameters):**

1. sqoop.env.mapreduce.job.queuename<=>-Dmapreduce.job.queuename
2. sqoop.args.connection.manager<===>--connection-manager
3. sqoop.args.connection.param.file<===>--connection-param-file
4. sqoop.args.driver<===>--driver
5. sqoop.args.hadoop.home<===>--hadoop-home
6. sqoop.args.hadoop.mapred.home<===>--hadoop-mapred-home
7. sqoop.args.help<===>help
8. sqoop.args.password<===>--password
9. sqoop.args.password.alias<===>--password-alias
10. sqoop.args.password.file<===>--password-file
11. sqoop.args.relaxed.isolation<===>--relaxed-isolation
12. sqoop.args.skip.dist.cache<===>--skip-dist-cache
13. sqoop.args.username<===>--username
14. sqoop.args.verbose<===>--verbose
15. sqoop.args.append<===>--append
16. sqoop.args.as.avrodatafile<===>--as-avrodatafile
17. sqoop.args.as.parquetfile<===>--as-parquetfile
18. sqoop.args.as.sequencefile<===>--as-sequencefile
19. sqoop.args.as.textfile<===>--as-textfile
20. sqoop.args.autoreset.to.one.mapper<===>--autoreset-to-one-mapper
21. sqoop.args.boundary.query<===>--boundary-query
22. sqoop.args.case.insensitive<===>--case-insensitive
23. sqoop.args.columns<===>--columns
24. sqoop.args.compression.codec<===>--compression-codec
25. sqoop.args.delete.target.dir<===>--delete-target-dir
26. sqoop.args.direct<===>--direct
27. sqoop.args.direct.split.size<===>--direct-split-size
28. sqoop.args.query<===>--query
29. sqoop.args.fetch.size<===>--fetch-size
30. sqoop.args.inline.lob.limit<===>--inline-lob-limit
31. sqoop.args.num.mappers<===>--num-mappers
32. sqoop.args.mapreduce.job.name<===>--mapreduce-job-name
33. sqoop.args.merge.key<===>--merge-key
34. sqoop.args.split.by<===>--split-by
35. sqoop.args.table<===>--table
36. sqoop.args.target.dir<===>--target-dir
37. sqoop.args.validate<===>--validate
38. sqoop.args.validation.failurehandler<===>--validation-failurehandler
39. sqoop.args.validation.threshold<===> --validation-threshold
40. sqoop.args.validator<===>--validator
41. sqoop.args.warehouse.dir<===>--warehouse-dir
42. sqoop.args.where<===>--where
43. sqoop.args.compress<===>--compress
44. sqoop.args.check.column<===>--check-column
45. sqoop.args.incremental<===>--incremental
46. sqoop.args.last.value<===>--last-value
47. sqoop.args.enclosed.by<===>--enclosed-by
48. sqoop.args.escaped.by<===>--escaped-by
49. sqoop.args.fields.terminated.by<===>--fields-terminated-by
50. sqoop.args.lines.terminated.by<===>--lines-terminated-by
51. sqoop.args.mysql.delimiters<===>--mysql-delimiters
52. sqoop.args.optionally.enclosed.by<===>--optionally-enclosed-by
53. sqoop.args.input.enclosed.by<===>--input-enclosed-by
54. sqoop.args.input.escaped.by<===>--input-escaped-by
55. sqoop.args.input.fields.terminated.by<===>--input-fields-terminated-by
56. sqoop.args.input.lines.terminated.by<===>--input-lines-terminated-by
57. sqoop.args.input.optionally.enclosed.by<===>--input-optionally-enclosed-by
58. sqoop.args.create.hive.table<===>--create-hive-table
59. sqoop.args.hive.delims.replacement<===>--hive-delims-replacement
60. sqoop.args.hive.database<===>--hive-database
61. sqoop.args.hive.drop.import.delims<===>--hive-drop-import-delims
62. sqoop.args.hive.home<===>--hive-home
63. sqoop.args.hive.import<===>--hive-import
64. sqoop.args.hive.overwrite<===>--hive-overwrite
65. sqoop.args.hive.partition.value<===>--hive-partition-value
66. sqoop.args.hive.table<===>--hive-table
67. sqoop.args.column.family<===>--column-family
68. sqoop.args.hbase.bulkload<===>--hbase-bulkload
69. sqoop.args.hbase.create.table<===>--hbase-create-table
70. sqoop.args.hbase.row.key<===>--hbase-row-key
71. sqoop.args.hbase.table<===>--hbase-table
72. sqoop.args.hcatalog.database<===>--hcatalog-database
73. sqoop.args.hcatalog.home<===>--hcatalog-home
74. sqoop.args.hcatalog.partition.keys<===>--hcatalog-partition-keys
75. sqoop.args.hcatalog.partition.values<===>--hcatalog-partition-values
76. sqoop.args.hcatalog.table<===>--hcatalog-table
77. sqoop.args.hive.partition.key<===>--hive-partition-key
78. sqoop.args.map.column.hive<===>--map-column-hive
79. sqoop.args.create.hcatalog.table<===>--create-hcatalog-table
80. sqoop.args.hcatalog.storage.stanza<===>--hcatalog-storage-stanza
81. sqoop.args.accumulo.batch.size<===>--accumulo-batch-size
82. sqoop.args.accumulo.column.family<===>--accumulo-column-family
83. sqoop.args.accumulo.create.table<===>--accumulo-create-table
84. sqoop.args.accumulo.instance<===>--accumulo-instance
85. sqoop.args.accumulo.max.latency<===>--accumulo-max-latency
86. sqoop.args.accumulo.password<===>--accumulo-password
87. sqoop.args.accumulo.row.key<===>--accumulo-row-key
88. sqoop.args.accumulo.table<===>--accumulo-table
89. sqoop.args.accumulo.user<===>--accumulo-user
90. sqoop.args.accumulo.visibility<===>--accumulo-visibility
91. sqoop.args.accumulo.zookeepers<===>--accumulo-zookeepers
92. sqoop.args.bindir<===>--bindir
93. sqoop.args.class.name<===>--class-name
94. sqoop.args.input.null.non.string<===>--input-null-non-string
95. sqoop.args.input.null.string<===>--input-null-string
96. sqoop.args.jar.file<===>--jar-file
97. sqoop.args.map.column.java<===>--map-column-java
98. sqoop.args.null.non.string<===>--null-non-string
99. sqoop.args.null.string<===>--null-string
100. sqoop.args.outdir<===>--outdir
101. sqoop.args.package.name<===>--package-name
102. sqoop.args.conf<===>-conf
103. sqoop.args.D<===>-D
104. sqoop.args.fs<===>-fs
105. sqoop.args.jt<===>-jt
106. sqoop.args.files<===>-files
107. sqoop.args.libjars<===>-libjars
108. sqoop.args.archives<===>-archives
109. sqoop.args.update.key<===>--update-key
110. sqoop.args.update.mode<===>--update-mode
111. sqoop.args.export.dir<===>--export-dir