Spark操作Kudu dataFrame操作kudu

一、DataFrameApi读取kudu表中的数据

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

• 代码示例

  

   

    

     

    
    

     

      /**
     
    
   

    

     

    
    

     

       * 使用DataFrameApi读取kudu表中的数据
     
    
   

    

     

    
    

     

       * @param sparkSession
     
    
   

    

     

    
    

     

       * @param kuduMaster
     
    
   

    

     

    
    

     

       * @param tableName
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     

      def getTableData(sparkSession: SparkSession, kuduMaster: String, tableName: String): Unit = {
     
    
   

    

     

    
    

     
  //定义map集合，封装kudu的master地址和要读取的表名
     
    
   

    

     

    
    

     
  val options = Map(
     
    
   

    

     

    
    

     
    "kudu.master" -> kuduMaster,
     
    
   

    

     

    
    

     
    "kudu.table" -> tableName
     
    
   

    

     

    
    

     

        )
     
    
   

    

     

    
    

     

        sparkSession.read.options(options).kudu.show()
     
    
   

    

     

    
    

     

      }
     
    
  
 

二、 DataFrameApi写数据到kudu表中

在通过DataFrame API编写时，目前只支持一种模式“append”。尚未实现的“覆盖”模式。

• 代码示例

  

   

    

     

    
    

     

      /**
     
    
   

    

     

    
    

     

       * 6）DataFrameApi写数据到kudu表中
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     

      def dataFrame2Kudu(session: SparkSession, kuduContext: KuduContext): Unit ={
     
    
   

    

     

    
    

     
  val data = List(person(3, "canglaoshi", 14, 0), person(4, "xiaowang", 18, 1))
     
    
   

    

     

    
    

     
  import  session.implicits._
     
    
   

    

     

    
    

     
  val dataFrame = data.toDF
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
  //目前，在kudu中，数据的写入只支持append追加
     
    
   

    

     

    
    

     

        dataFrame.write.mode("append").options(kuduOptions).kudu
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
  //查看结果
     
    
   

    

     

    
    

     
  //导包
     
    
   

    

     

    
    

     
  import org.apache.kudu.spark.kudu._
     
    
   

    

     

    
    

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

    

     

    
    

     

        sparkSession.read.options(kuduOptions).kudu.show()
     
    
   

    

     

    
    

     

      }
     
    
  
 

三、​​​​​​​使用sparksql操作kudu表

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

• 代码示例

  

   

    

     

    
    

     

      /**
     
    
   

    

     

    
    

     

       * 使用sparksql操作kudu表
     
    
   

    

     

    
    

     

       * @param sparkSession
     
    
   

    

     

    
    

     

       * @param sc
     
    
   

    

     

    
    

     

       * @param kuduMaster
     
    
   

    

     

    
    

     

       * @param tableName
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     

      def SparkSql2Kudu(sparkSession: SparkSession, sc: SparkContext, kuduMaster: String, tableName: String): Unit = {
     
    
   

    

     

    
    

     
  //定义map集合，封装kudu的master地址和表名
     
    
   

    

     

    
    

     
  val options = Map(
     
    
   

    

     

    
    

     
    "kudu.master" -> kuduMaster,
     
    
   

    

     

    
    

     
    "kudu.table" -> tableName
     
    
   

    

     

    
    

     

        )
     
    
   

    

     

    
    

     
  val data = List(persont(10, "小张", 30, 0), person(11, "小王", 40, 0))
     
    
   

    

     

    
    

     
  import sparkSession.implicits._
     
    
   

    

     

    
    

     

        val dataFrame: DataFrame = sc.parallelize(data).toDF
     
    
   

    

     

    
    

     
  //把dataFrame注册成一张表
     
    
   

    

     

    
    

     

        dataFrame.createTempView("temp1")
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
  //获取kudu表中的数据，然后注册成一张表
     
    
   

    

     

    
    

     

        sparkSession.read.options(options).kudu.createTempView("temp2")
     
    
   

    

     

    
    

     
  //使用sparkSQL的insert操作插入数据
     
    
   

    

     

    
    

     

        sparkSession.sql("insert into table temp2 select * from temp1")
     
    
   

    

     

    
    

     

        sparkSession.sql("select * from temp2 where age >30").show()
     
    
   

    

     

    
    

     

      }
     
    
  
 

• 📢博客主页：https://lansonli.blog.csdn.net
• 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
• 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
• 📢大数据系列文章会每天更新，停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨

文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。

原文链接：lansonli.blog.csdn.net/article/details/123030789