spark操作Kudu之读 - 使用DataFrame API

时间:2022-04-24 08:48:38

spark操作Kudu之读 - 使用DataFrame API

虽然我们可以通过上面显示的KuduContext执行大量操作,但我们还可以直接从默认数据源本身调用读/写API。

要设置读取,我们需要为Kudu表指定选项,命名我们要读取的表以及为表提供服务的Kudu集群的Kudu主服务器列表。

import org.apache.kudu.spark.kudu._
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession /**
* Created by angel;
*/
object DataFrame_read {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("AcctfileProcess")
//设置Master_IP并设置spark参数
.setMaster("local")
.set("spark.worker.timeout", "500")
.set("spark.cores.max", "10")
.set("spark.rpc.askTimeout", "600s")
.set("spark.network.timeout", "600s")
.set("spark.task.maxFailures", "1")
.set("spark.speculationfalse", "false")
.set("spark.driver.allowMultipleContexts", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val sparkContext = SparkContext.getOrCreate(sparkConf)
val sqlContext = SparkSession.builder().config(sparkConf).getOrCreate().sqlContext
//TODO 1:定义表名
val kuduTableName = "spark_kudu_tbl"
val kuduMasters = "hadoop01:7051,hadoop02:7051,hadoop03:7051"
//使用spark创建kudu表
val kuduContext = new KuduContext(kuduTableName, sqlContext.sparkContext) //TODO 2:配置kudu参数
val kuduOptions: Map[String, String] = Map(
"kudu.table" -> kuduTableName,
"kudu.master" -> kuduMasters)
//TODO 3:执行读取操作
val customerReadDF = sqlContext.read.options(kuduOptions).kudu
val filterData = customerReadDF.select("name" ,"age", "city").filter("age<30")
//TODO 4:打印
filterData.show()
}
}