需求:
1、在工作中，我们常常需要将外部 hive 或者 mysql、oracle 等数据源导入到clickhouse中，对于多种外部数据源，是否有通用的数据导入方案?

2、我们在clickhouse上维持一张查询主表，但外部数据源表是hive增量表，新增数据需要同步更新到clickhouse上，是否有不通过第三方组件的插入方式？

解决:
1、外部数据源有可能是多种，如下图：

那么这里不纠结某个外部数据源的导入速度，我这里提供一种通用方式，可以先在clickhouse中
创建一张外部引擎表，详细官网都有格式介绍：https://clickhouse.com/docs/zh/engines/table-engines/integrations/hdfs/

由于clickhouse支持insert into xxx select * from xxx 语句进行写入

故我们可以将外部引擎表通过此函数导入进需要经常查询的MergeTree引擎表中

那么有人可能会说：既然外部引擎表可以供我们查询，我们为何要重新写入到MergeTree引擎表？

这是因为clickhouse在OLAP查询分析中的核心引擎就是MergeTree，而外部表只是嫁接表，无法使用clickhouse的核心能力，故需要有一层导入过程，这里可以将外部引擎表想象成中间层，最终供外部查询的数据都应该在MergeTree表中，如下图：

2、在hive增量表层面上，新增数据大部分会存储到一张临时表然后合并到hive主表，亦或者以文件的方式存储再转入至hive目录下，针对这种新增数据很明确的情况，我们常常使用的方式是通过clickhouse创建一张临时外部表指向hive临时表，然后通过 insert into xxx select * from xxx 语句写入进主表即可。

对于spark外部数据源来说，要先了解这几个类

BaseRelation:定义数据的schema信息，把我们的数据转成RDD[Row]
RelationProvider:是一个relation的提供者，创建BaseRelation
TableScan：读取数据并构建行，拿出所有的数据
PrunedScan:列裁剪的
PrunedFilteredScan：列裁剪➕过滤

InsertableRelation：回写数据的relation

insertableRelation有如下假设需要注意：
/**

• A BaseRelation that can be used to insert data into it through the insert method.
• If overwrite in insert method is true, the old data in the relation should be overwritten with
• the new data. If overwrite in insert method is false, the new data should be appended.
• InsertableRelation has the following three assumptions.
• 1. It assumes that the data (Rows in the DataFrame) provided to the insert method
• exactly matches the ordinal of fields in the schema of the BaseRelation.
• 2. It assumes that the schema of this relation will not be changed.
• Even if the insert method updates the schema (e.g. a relation of JSON or Parquet data may have a
• schema update after an insert operation), the new schema will not be used.
• 3. It assumes that fields of the data provided in the insert method are nullable.
• If a data source needs to check the actual nullability of a field, it needs to do it in the
• insert method.
• @since 1.3.0
  */
  @InterfaceStability.Stable
  trait InsertableRelation {
  def insert(data: DataFrame, overwrite: Boolean): Unit
  }

对于spark外部数据源来说，要先了解这几个类

BaseRelation:定义数据的schema信息，把我们的数据转成RDD[Row]
RelationProvider:是一个relation的提供者，创建BaseRelation
TableScan：读取数据并构建行，拿出所有的数据
PrunedScan:列裁剪的
PrunedFilteredScan：列裁剪➕过滤

InsertableRelation：回写数据的relation

insertableRelation有如下假设需要注意：
/**

• A BaseRelation that can be used to insert data into it through the insert method.
• If overwrite in insert method is true, the old data in the relation should be overwritten with
• the new data. If overwrite in insert method is false, the new data should be appended.
• InsertableRelation has the following three assumptions.
• 1. It assumes that the data (Rows in the DataFrame) provided to the insert method
• exactly matches the ordinal of fields in the schema of the BaseRelation.
• 2. It assumes that the schema of this relation will not be changed.
• Even if the insert method updates the schema (e.g. a relation of JSON or Parquet data may have a
• schema update after an insert operation), the new schema will not be used.
• 3. It assumes that fields of the data provided in the insert method are nullable.
• If a data source needs to check the actual nullability of a field, it needs to do it in the
• insert method.
• @since 1.3.0
  */
  @InterfaceStability.Stable
  trait InsertableRelation {
  def insert(data: DataFrame, overwrite: Boolean): Unit
  }

下面我们可以简单看一下JDBCRelation

private[sql] case class JDBCRelation(
parts: Array[Partition], jdbcOptions: JDBCOptions)(@transient val sparkSession: SparkSession)
extends BaseRelation
with PrunedFilteredScan
with InsertableRelation

继承了 BaseRelation ，BaseRelation 是一个抽象类，就需要实现他的属性和方法：
override def sqlContext: SQLContext = sparkSession.sqlContext

override val needConversion: Boolean = false

override val schema: StructType = {
val tableSchema = JDBCRDD.resolveTable(jdbcOptions)
jdbcOptions.customSchema match {
case Some(customSchema) => JdbcUtils.getCustomSchema(
tableSchema, customSchema, sparkSession.sessionState.conf.resolver)
case None => tableSchema
}
}
…

import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.sources.{BaseRelation, RelationProvider, SchemaRelationProvider}
import org.apache.spark.sql.types.StructType

/**
  * @Author: lih
  * @Date: 2019/8/2 11:40 PM
  * @Version 1.0
  */
class DefaultSource extends RelationProvider with SchemaRelationProvider{
  override def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation = {
    createRelation(sqlContext,parameters,null)
  }

  override def createRelation(sqlContext: SQLContext, parameters: Map[String, String], schema: StructType): BaseRelation = {
    val path = parameters.get("path")

    path match {
      case Some(p)=> new TextDataSourceRelation(sqlContext,p,schema)
      case _=> throw new IllegalArgumentException("path is required ...")
    }


  }
}

import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._

/**
  * @Author: lih
  * @Date: 2019/8/2 11:41 PM
  * @Version 1.0
  */
class TextDataSourceRelation(override val sqlContext: SQLContext,
                             path: String,
                             userSchema: StructType
                            ) extends BaseRelation with TableScan with Logging with Serializable {

  override def schema: StructType = {
    if (userSchema != null) {
      userSchema
    } else {
      StructType(
        StructField("id", LongType, false) ::
          StructField("name", StringType, false) ::
          StructField("gender", StringType, false) ::
          StructField("salar", LongType, false) ::
          StructField("comm", LongType, false) :: Nil
      )
    }
  }


  override def buildScan(): RDD[Row] = {
    logError("this is ruozedata custom buildScan...")

    var rdd = sqlContext.sparkContext.wholeTextFiles(path).map(_._2)
    val schemaField = schema.fields

    // rdd + schemaField
    val rows = rdd.map(fileContent => {
      val lines = fileContent.split("\n")
      val data = lines.map(_.split(",").map(x=>x.trim)).toSeq

      val result = data.map(x => x.zipWithIndex.map{
        case  (value, index) => {

          val columnName = schemaField(index).name

           caseTo(if(columnName.equalsIgnoreCase("gender")) {
            if(value == "0") {
              "男"
            } else if(value == "1"){
              "女"
            } else {
              "未知"
            }
          } else {
            value
          }, schemaField(index).dataType)
        }
      })
      result.map(x => Row.fromSeq(x))
    })

    rows.flatMap(x=>x)
  }


  def caseTo(value: String, dataType: DataType) = {
      dataType match {
        case _:DoubleType => value.toDouble
        case _:LongType => value.toLong
        case _:StringType => value
    }
  }
}

数据：
1,li,0,100000,2000
2,zhang,1,20000,23223

object Test {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName).master("local[8]")
      .getOrCreate()


    val df = spark
      .read
      .format("com.ztgx.datasourse.DefaultSource")
      .load("file:///Users/mac/Desktop/1.txt")

    df.show()

    spark.stop()
  }

}

结果：

+---+-----+------+------+-----+
| id| name|gender| salar| comm|
+---+-----+------+------+-----+
|  1|   li|     男|100000| 2000|
|  2|zhang|     女| 20000|23223|
+---+-----+------+------+-----+

产生背景

概述

目标

操作parquet文件数据

操作hive数据

操作mysql数据

统一

产生背景

每一个spark都是以加载数据开始，经过一系列处理，最后存储到其他地方；

不同格式，不同压缩格式，不同存储接口，用户肯定希望从不同数据源收集数据

方便、快速从不同数据源（）经过混合处理（json直接和parqent jion）再将结果以特定格式写回到指定系统上去

sparksql 1.2====》外部数据源api

问题：

1.加载、保存数据并不简单，比如从关系数据库sqoop加载到hdfs然后..

2.解析原生数据（text/json/parquet）

3.转换数据格式

数据集存储在不同的存储系统、格式上面

api概述：

一种扩展方式，将外部数据源整合到sparl sql中

读写各种格式（指定格式和路径local,分布式）

目标：

开发人员：是否需要把代码合并到spark源码中？比如访问微博（按照dataframe api实现）

对于使用人员：非常容易加载保存

支持：build-in json parquet jdbc 其他数据源：packages:外部的，非spark内置的，一个网站

操作 parqeut文件数据