客快物流大数据项目(四十六):Spark操作Kudu dataFrame操作kudu

举报
Lansonli 发表于 2022/02/24 00:38:01 2022/02/24
4k+ 0 0
【摘要】 Spark操作Kudu dataFrame操作kudu 一、DataFrameApi读取kudu表中的数据 虽然我们可以通过上面显示的KuduContext执行大量操作,但我们还可以直接从默认数据源本身调用读/写API。要设置读取,我们需要为Kudu表指定选项,命名我们要读取的表以及为表提供服务的Kudu集群的Kudu主服务器列...

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

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

作者其他文章

评论(0

抱歉,系统识别当前为高风险访问,暂不支持该操作

    全部回复

    上滑加载中

    设置昵称

    在此一键设置昵称,即可参与社区互动!

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

    *长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。