DataFrame API操作Kudu

  虽然我们可以通过 KuduContext 执行大量操作，但我们还可以直接从默认数据源本身调用读/写 API。要设置读取，我们需要为 Kudu 表指定选项，命名我们要读取的表以及为表提供服务的 Kudu 集群的 Kudu 主服务器列表。

1.1、DataFrame操作kudu 1.1.1、DataFrame 读取kudu

  @Test
  def query(): Unit = {
    // 1、创建KuduContext 和SparkSession
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("sparkOpKudu")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    // 2、创建kuduContext
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    val tableName = "sparkopKudu"

    //定义 map 集合，封装 kudu 的 master 地址和要读取的表名
    val options = Map(
      "kudu.master" -> KUDU_MASTER,
      "kudu.table" -> tableName
    )

    val df = spark.read.options(options)
      .format("kudu").load
    df.show()
  }

1.1.1.1、ClassNotFoundException: kudu.DefaultSource

则请确认引入的jar包版本（用1.9.0版本，更早版本会报错）

或者使用cloudera的库

 org.apache.kudu
 kudu-client-tools
 1.6.0-cdh5.14.0
 

1.1.2、DataFrame写入Kudu

  @Test
  def DFWrite2Kudu(): Unit = {
    // 1、创建KuduContext 和SparkSession
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("sparkOpKudu")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    // 2、创建kuduContext
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    val tableName = "sparkopKudu"

    //定义 map 集合，封装 kudu 的 master 地址和要读取的表名
    val options = Map(
      "kudu.master" -> KUDU_MASTER,
      "kudu.table" -> tableName
    )

    import spark.implicits._
    val data = List(Person(7, "qiqi", 30), Person(8, "xiaoba", 40))
    //导包
    import org.apache.kudu.spark.kudu._
    val df = data.toDF

    //把 dataFrame 结果写入到 kudu 表中 ,目前只支持 append 追加
    df.write.options(options).mode("append").kudu


    //加载表的数据，导包调用 kudu 方法，转换为 dataFrame，最后在使用 show 方法显示结果
    spark.read.options(options).kudu.show()
  }

1.2、Spark SQL 操作Kudu

可以选择使用 Spark SQL 直接使用INSERT语句写入Kudu 表；与’append’类 似，INSERT语句实际上将默认使用UPSERT 语义处理.

  @Test
  def sparksql2Kudu(): Unit = {
    // 1、创建KuduContext 和SparkSession
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("sparkOpKudu")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    // 2、创建kuduContext
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    val tableName = "sparkopKudu"

    //定义 map 集合，封装 kudu 的 master 地址和要读取的表名
    val options = Map(
      "kudu.master" -> KUDU_MASTER,
      "kudu.table" -> tableName
    )

    val data = List(Person(10, "小张", 30), Person(11, "小王", 40))
    import spark.implicits._
    val df: DataFrame = spark.sparkContext.parallelize(data).toDF

    //把 dataFrame 注册成一张表
    df.createTempView("temp1")
    //获取 kudu 表中的数据，然后注册成一张表
    import org.apache.kudu.spark.kudu._
    spark.read.options(options).kudu.createTempView("temp2")
    //使用 sparkSQL 的 insert 操作插入数据
    spark.sql("insert into table temp2 select * from temp1")
    spark.sql("select * from temp2 where age >30").show()


    //加载表的数据，导包调用 kudu 方法，转换为 dataFrame，最后在使用 show 方法显示结果
    spark.read.options(options).kudu.show()
  }

1.3、Kudu native RDD

Spark 与 Kudu 的集成同时提供了 kudu RDD.

  @Test
  def kuduNativeRDD (): Unit = {
    // 1、创建KuduContext 和SparkSession
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("sparkOpKudu")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    // 2、创建kuduContext
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    val kuduContext = new KuduContext(KUDU_MASTER, spark.sparkContext)

    //使用 kuduContext 对象调用 kuduRDD 方法，需要 sparkContext 对象，表名，想要的字段名称
    val kuduRDD: RDD[Row] = kuduContext.kuduRDD(spark.sparkContext,"sparkopKudu",Seq("name","age"))
    //操作该 rdd 打印输出
    val result: RDD[(String, Int)] = kuduRDD.map {
      case Row(name: String, age: Int) => (name, age)
    }
    result.foreach(println)
  }