开发指南

docker load < sparkee-builder.tar

/**
  * 此应用实现将任务第一笔输入指定的 MySQL 表转储到第一笔输出指定的 MySQL 表。
  */
package com.woa.git.xiangminli.helloworld

import java.util.Properties

import org.apache.spark.sql.{SaveMode, SparkSession}

object HelloWorldMySQL extends App {

  val ss = SparkSession
    .builder()
    .appName("Hello World with MySQL by XML")
    .getOrCreate()
  
  // 获取任务输入/输出参数组成的 TaskParams 结构
  // TaskParams 的详情参见 《3.1. 核心数据结构》
  val params = ss.sparkContext.getConf.teeConf.get.taskParams()

  // 打印输入配置信息用于调试
  for (v <- params.inputs) {
    println(
      s"- uri: ${v.uri.getUri()}, name: ${v.uri.getName()}, token: '${v.authToken}'",
    )

    // 强制要求指定数据源 CA 证书
    val certPem = v.caCertChainPem.getOrElse(
      throw new IllegalStateException("miss input trust cert key store"),
    )
    println(s"  CA:\\n${certPem}")
  }

  assert(!params.inputs.isEmpty, "miss inputs")

  // 解析数据库地址和表名
  val (dsn, tableName) = {
    val uri = params.inputs(0).uri
    (s"jdbc:${uri.getUri()}", uri.getName())
  }
  println(s"dsn: $dsn, table: $tableName")

  // 从第一笔输入的配置信息解析出账号、密码和 MySQL 服务端的 CA 证书
  // CA 证书用于建立 TLS 链接过程验证 MySQL 服务证书的合法性
  val (user, password, inTrustCertKeyStore) = {
    val input = params.inputs(0)
    val token = input.authToken.split(":") // 假设 authToken 的形式为 "用户名：密码"
    if (token.length != 2) {
      throw new IllegalStateException(s"${input.authToken} is malformed")
    }

    (token(0), token(1), input.trustCertKeyStore.get)
  }
  // 打印信息用于调试。发布的应用应删除打印这些敏感信息的打印语句。
  println(s"user: '$user', password: '$password'")
  println(
    s"trustCertificateKeyStoreUrl: '${inTrustCertKeyStore.url}'," +
      s" trustCertificateKeyStorePassword: '${inTrustCertKeyStore.password}'",
  )

  val props = new Properties
  props.put("user", user)
  props.put("password", password)
  // 添加验证服务端 TLS 通信配置
  // @warn 此步骤对保护数据源安全至关重要，能够有效抵御中间人攻击
  props.put("sslMode", "VERIFY_CA")
  // 设置服务端 JKS 格式 CA 证书路径
  // @warn 此步骤对保护数据源安全至关重要，能够有效抵御中间人攻击
  props.put("trustCertificateKeyStoreUrl", inTrustCertKeyStore.url)
  // 设置解密服务端 CA 证书的口令
  props.put("trustCertificateKeyStorePassword", inTrustCertKeyStore.password)

  // 从数据库读取输入
  val inDF = ss.read.jdbc(dsn, tableName, props)

  // 按需对 inDF 执行具体计算
  // 例如，打印 inDF 对应的表结构信息
  inDF.printSchema()
  // 例如，打印 inDF 对应表的前 3 行
  inDF.show(3)

  // 还可以调用 DataFrame 支持的所有 API 对 inDF 进行其他转换
  // 具体 API 参见这里
  //   https://spark.apache.org/docs/3.1.3/api/scala/org/apache/spark/sql/Dataset.html 
  // 在此仅简单地将 inDF 复制给 outDF
  val outDF = inDF

  // 获取第一笔输出的配置信息 Output
  val out = params.outputs(0)

  // 构造结果输出地址
  val outUrl = s"jdbc:${out.uri}"
  println(s"save output to $outUrl in table ${out.name}")

  // 将结果写出至具体数据库
  // @note：此处假设结果和输入采用相同的账号和密码。正常情况下，应执行和解析输入数据库账号和密码相同的逻辑从数据库解析出账号和
  // 密码用于读写数据库。
  outDF.write.mode(SaveMode.Overwrite).jdbc(outUrl, out.name, props)

  println("done :)")
}


case class TaskParams(inputs: Array[Input], outputs: Array[Output])

case class Input(
    uri: ExtendedURI,
    authToken: String,
    seq: Array[Byte],
    caCertChainPem: Option[String],
    ...
) {
  /**
   * JKS 编码 caCertChainPem 所得可信证书链
   */
  val trustCertKeyStore: TrustCertKeyStore = ...
}

case class Output(
    uri: String,
    authToken: String,
    name: String,
    seq: Array[Byte],
    caCertChainPem: Option[String],
    ...
) {
  /**
   * JKS 编码 caCertChainPem 所得可信证书链
   */
  val trustCertKeyStore: TrustCertKeyStore = ...
}

case class TrustCertKeyStore(certChainPem: String, password: String, mType: String = "JKS") {
  val url: String = ...
}