Sqoop Engine usage documentation

    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 环境配置清单

    Linkis Parameter NameParameter ContentRemark
    wds.linkis.hadoop.site.xmlSet sqoop to load hadoop parameter file locationRequired，Reference example：”/etc/hadoop/conf/core-site.xml;/etc/hadoop/conf/hdfs-site.xml;/etc/hadoop/conf/yarn-site.xml;/etc/hadoop/conf/mapred-site.xml”
    sqoop.fetch.status.intervalSet the interval time for obtaining sqoop execution statusNot required, the default value is 5s

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

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

    and then deploy to

    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

    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

    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.      .setStartupParams(startUpMap)
    31.      .setMaxSubmitTime(30000)
    32.      .addExecuteUser("freeuser")
    33.   val onceJob = importJob(builder)
    35.   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 Monitor: ${operator.apply()}")
    64.         println(s"Progress: ${progressOperator.apply()}")
    65.       }
    66.     })
    67.   }
    68.   onceJob.waitForCompleted()
    69.   println(onceJob.getStatus)
    71.   System.exit(0)
    76.    def importJob(jobBuilder: SimpleOnceJobBuilder): SubmittableSimpleOnceJob = {
    77.      jobBuilder
    78.        .addJobContent("sqoop.env.mapreduce.job.queuename", "queue_10")
    79.        .addJobContent("sqoop.mode", "import")
    80.        .addJobContent("sqoop.args.connect", "jdbc:mysql://127.0.0.1:3306/exchangis")
    81.        .addJobContent("sqoop.args.username", "free")
    82.        .addJobContent("sqoop.args.password", "testpwd")
    83.        .addJobContent("sqoop.args.query", "select id as order_number, sno as time from" +
    84.          " exchangis where sno =1 and $CONDITIONS")
    85.        .addJobContent("sqoop.args.hcatalog.database", "freedb")
    86.        .addJobContent("sqoop.args.hcatalog.table", "zy_test")
    87.        .addJobContent("sqoop.args.hcatalog.partition.keys", "month")
    88.        .addJobContent("sqoop.args.hcatalog.partition.values", "3")
    89.        .addJobContent("sqoop.args.num.mappers", "1")
    90.        .build()
    91.    }
    94.    def exportJob(jobBuilder: SimpleOnceJobBuilder): SubmittableSimpleOnceJob = {
    95.       jobBuilder
    96.         .addJobContent("sqoop.env.mapreduce.job.queuename", "queue1")
    97.         .addJobContent("sqoop.mode", "import")
    98.         .addJobContent("sqoop.args.connect", "jdbc:mysql://127.0.0.1:3306/exchangis")
    99.         .addJobContent("sqoop.args.query", "select id as order, sno as great_time from" +
    100.           " exchangis_table where sno =1 and $CONDITIONS")
    101.         .addJobContent("sqoop.args.hcatalog.database", "hadoop")
    102.         .addJobContent("sqoop.args.hcatalog.table", "partition_33")
    103.         .addJobContent("sqoop.args.hcatalog.partition.keys", "month")
    104.         .addJobContent("sqoop.args.hcatalog.partition.values", "4")
    105.         .addJobContent("sqoop.args.num.mappers", "1")