暑期2021项目经验分享：实现Spark对接openGauss

1. Spark DataSource接口简介

def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder

def build(): Scan

def readSchema(): StructType
def toBatch: Batch

def planInputPartitions(): Array[InputPartition] 
def createReaderFactory(): PartitionReaderFactory

def createReader(partition: InputPartition): PartitionReader[InternalRow]

def next : Boolean
def get : T
def close() : Unit

def newWriteBuilder(logicalWriteInfo: LogicalWriteInfo): WriteBuilder

def buildForBatch(): BatchWrite

def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo):DataWriterFactory
def commit(writerCommitMessages: Array[WriterCommitMessage]):Unit
def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit

commit: 正如我们前面所讨论的，上面的方法是作者事务支持的一部分。当所有的写操作完成后，调用这个方法来提交。这就是全面提交。单独的分区提交将出现在 DataWriter 接口中，我们将在下面讨论它。WriterCommitMessage 是一个消息接口，数据源应该使用它来定义自己的消息，以指示每个分区写的状态。

def createDataWriter(partitionId:Int, taskId:Long): DataWriter[T]

partitionId: 分区的 id。这有助于编写器理解它正在写入的分区。

taskId: task的id。方法返回 DataWriter。

最后我们有一个实际写入数据的接口：

def write(record: T): Unit
def commit(): WriterCommitMessage
def abort(): Unit

这个接口中由 commit 方法发送的 WriterCommitMessage 是发送给 BatchWrite 的消息。这有助于数据源理解每个分区的状态。

2.1.1 直接获取

从maven中央仓库获取

<groupId>org.opengauss</groupId>
<artifactId>opengauss-jdbc</artifactId>

2.2.1 本地包的使用

<!-- https://mvnrepository.com/artifact/org.opengauss/opengauss-jdbc -->
<dependency>
    <groupId>org.opengauss</groupId>
    <artifactId>opengauss-jdbc</artifactId>
    <version>2.0.1-compatibility</version>
</dependency>

3. 基于openGauss JDBC 实现DataSource接口

TableProvider 接口可以对现有表应用数据操作，比如读取、追加、删除和覆盖。

class DefaultSource extends TableProvider {
  override def inferSchema(options: CaseInsensitiveStringMap): StructType = OpenGaussTable.schema

  override def getTable(
                         schema: StructType,
                         partitioning: Array[Transform],
                         properties: util.Map[String, String]
                       ): Table = new OpenGaussTable(properties.get("tableName"))
}

实现 SupportsRead、SupportsWrite 接口

class OpenGaussTable(val name: String) extends SupportsRead with SupportsWrite {
  override def schema(): StructType = OpenGaussTable.schema

  override def capabilities(): util.Set[TableCapability] = Set(
    TableCapability.BATCH_READ,
    TableCapability.BATCH_WRITE
  ).asJava

  override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new OpenGaussScanBuilder(options)

  override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = new OpenGaussWriteBuilder(info.options)
}

object OpenGaussTable {
  /*Table products*/
  /*Database school, table course*/
  val schema: StructType = new StructType().add("cor_id", IntegerType).add("cor_name", StringType).add("cor_type", StringType).add("credit", DoubleType)
}

case class ConnectionProperties(url: String, user: String, password: String, tableName: String, partitionColumn: String, partitionSize: Int)

class OpenGaussScanBuilder(options: CaseInsensitiveStringMap) extends ScanBuilder {
  override def build(): Scan = new OpenGaussScan(ConnectionProperties(
    options.get("url"), options.get("user"), options.get("password"), options.get("tableName"), options.get("partitionColumn"), options.get("partitionSize").toInt
  ))
}

class OpenGaussPartition extends InputPartition

class OpenGaussScan(connectionProperties: ConnectionProperties) extends Scan with Batch {
  override def readSchema(): StructType = OpenGaussTable.schema

  override def toBatch: Batch = this

  override def planInputPartitions(): Array[InputPartition] = Array(new OpenGaussPartition)

  override def createReaderFactory(): PartitionReaderFactory = new OpenGaussPartitionReaderFactory(connectionProperties)
}

class OpenGaussPartitionReaderFactory(connectionProperties: ConnectionProperties)
  extends PartitionReaderFactory {
  override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new OpenGaussPartitionReader(connectionProperties)
}


class OpenGaussPartitionReader(connectionProperties: ConnectionProperties) extends PartitionReader[InternalRow] {
  private val connection = DriverManager.getConnection(
    connectionProperties.url, connectionProperties.user, connectionProperties.password
  )
  private val statement = connection.createStatement()
  private val resultSet = statement.executeQuery(s"select * from ${connectionProperties.tableName}")

  override def next(): Boolean = resultSet.next()

  override def get(): InternalRow = InternalRow(
    resultSet.getInt(1),
    UTF8String.fromString(resultSet.getString(2)),
    UTF8String.fromString(resultSet.getString(3)),
    resultSet.getDouble(4))

  override def close(): Unit = connection.close()

}

class OpenGaussWriteBuilder(options: CaseInsensitiveStringMap) extends WriteBuilder {
  override def buildForBatch(): BatchWrite = new OpenGaussBatchWrite(ConnectionProperties(
    options.get("url"), options.get("user"), options.get("password"), options.get("tableName"), options.get("partitionColumn"), options.get("partitionSize").toInt
  ))
}

class OpenGaussBatchWrite(connectionProperties: ConnectionProperties) extends BatchWrite {
  override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo): DataWriterFactory =
    new OpenGaussDataWriterFactory(connectionProperties)

  override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit = {}

  override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit = {}
}

class OpenGaussDataWriterFactory(connectionProperties: ConnectionProperties) extends DataWriterFactory {
  override def createWriter(partitionId: Int, taskId:Long): DataWriter[InternalRow] =
    new OpenGaussWriter(connectionProperties)
}

object WriteSucceeded extends WriterCommitMessage

class OpenGaussWriter(connectionProperties: ConnectionProperties) extends DataWriter[InternalRow] {

  val connection = DriverManager.getConnection(
    connectionProperties.url,
    connectionProperties.user,
    connectionProperties.password
  )

  val statement = "insert into ${connectionProperties.tableName} (cor_name, cor_type, credit) values (?,?,?)"
  val preparedStatement = connection.prepareStatement(statement)

  override def write(record: InternalRow): Unit = {
    val cor_name = record.getString(0)
    val cor_type = record.getString(1)
    val credit = record.getDouble(2)

    preparedStatement.setString(0, cor_name)
    preparedStatement.setString(1, cor_type)
    preparedStatement.setDouble(2, credit)
    preparedStatement.executeUpdate()
  }

  override def commit(): WriterCommitMessage = WriteSucceeded

  override def abort(): Unit = {}

  override def close(): Unit = connection.close()
}

直接使用 JDBC 接口则是（这样使用则不是V2）：

作为学生，也需要平衡时间来参加暑期2021，在这个过程中，我对 Spark 对接 openGauss 的流程做了验证。在整个流程的验证中发现了一些问题，也在社区提交了 Issue 和 PR 进行解决，例如：

其实我在暑假期间同时参加了 openGauss 社区的任务打榜赛，即根据个人的issue和 PR 数量排名。在参加这两个项目的过程中，社区的工作人员都是非常和善的，我碰到的问题也能及时得到回应和解决，提交的PR也会及时被Review。在参加打榜赛的时候，我发现了一些规则上不完善的问题，反馈之后工作人员也及时更新了规则。

参与暑期2021也让我对开源技术有了更深的理解。Git 推动了开源文化的发展并改善了开发者的工作模式，但是除了技术之外，社区参与者的积极性与负责程度决定了开源社区的氛围。导师与社区中帮助我的人让我认识到开源文化中，人是最重要的组成部分。作为一名个人开发者，热情和积极性是很重要的动力，在开源社区中所付出的东西还会有很多额外收获，包括但不限于：提升专业技能、遇见对你有帮助的大牛、提高语言和组织能力清晰的描述问题与解决方案、积累个人声望提高存在感。在这个过程中，我确实在这几个方面都得到了提高。