客快物流大数据项目(四十四)：Spark操作Kudu创建表

Lansonli

发布于 2022-02-24 15:07:33

5250

发布于 2022-02-24 15:07:33

文章被收录于专栏：Lansonli技术博客Lansonli技术博客

Spark操作Kudu创建表

Spark与KUDU集成支持：
- DDL操作（创建/删除）
- 本地Kudu RDD
- Native Kudu数据源，用于DataFrame集成
- 从kudu读取数据
- 从Kudu执行插入/更新/ upsert /删除
- 谓词下推
- Kudu和Spark SQL之间的模式映射
- 到目前为止，我们已经听说过几个上下文，例如SparkContext，SQLContext，HiveContext， SparkSession，现在，我们将使用Kudu引入一个KuduContext。这是可以在Spark应用程序中广播的主要可序列化对象。此类代表在Spark执行程序中与Kudu Java客户端进行交互。
- KuduContext提供执行DDL操作所需的方法，与本机Kudu RDD的接口，对数据执行更新/插入/删除，将数据类型从Kudu转换为Spark等。

创建表

定义kudu的表需要分成5个步骤：
- 提供表名
- 提供schema
- 提供主键
- 定义重要选项；例如：定义分区的schema
- 调用create Table api
代码开发

package cn.it

import java.util
import cn.it.SparkKuduDemo.TABLE_NAME
import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object SparkKuduTest {
  def main(args: Array[String]): Unit = {
    //构建sparkConf对象
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkKuduTest").setMaster("local[2]")

    //构建SparkSession对象
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    //获取sparkContext对象
    val sc: SparkContext = sparkSession.sparkContext
    sc.setLogLevel("warn")

    //构建KuduContext对象
    val kuduContext = new KuduContext("node2:7051", sc)

    //1.创建表操作
    createTable(kuduContext)

    /**
     * 创建表
     *
     * @param kuduContext
     * @return
     */
    def createTable(kuduContext: KuduContext) = {
      //如果表不存在就去创建
      if (!kuduContext.tableExists(TABLE_NAME)) {

        //构建创建表的表结构信息,就是定义表的字段和类型
        val schema: StructType = StructType(
          StructField("userId", StringType, false) ::
            StructField("name", StringType, false) ::
            StructField("age", IntegerType, false) ::
            StructField("sex", StringType, false) :: Nil)

        //指定表的主键字段
        val keys = List("userId")

        //指定创建表所需要的相关属性
        val options: CreateTableOptions = new CreateTableOptions
        //定义分区的字段
        val partitionList = new util.ArrayList[String]
        partitionList.add("userId")
        //添加分区方式为hash分区
        options.addHashPartitions(partitionList, 6)

        //创建表
        kuduContext.createTable(TABLE_NAME, schema, keys, options)
      }
    }
  }
}

定义表时要注意的是Kudu表选项值。你会注意到在指定组成范围分区列的列名列表时我们调用“asJava”方法。这是因为在这里，我们调用了Kudu Java客户端本身，它需要Java对象（即java.util.List）而不是Scala的List对象；（要使“asJava”方法可用，请记住导入JavaConverters库。）创建表后，通过将浏览器指向http//master主机名:8051/tables