說明
OSS Select還支持其他過濾條件，詳情請(qǐng)參見 SelectObject 。
對(duì)比TPC-H的查詢通過測(cè)試TPC-H中query1.sql對(duì)于lineitem這個(gè)table的查詢性能，來檢驗(yàn)配置效果 。為了能使OSS Select過濾更多的數(shù)據(jù)，我們將where條件由l_shipdate <= ‘1998-09-16’改為where l_shipdate > ‘1997-09-16’，測(cè)試數(shù)據(jù)大小為2.27 GB 。僅使用Spark SQL查詢和在Spark SQL上使用OSS Select查詢的方式如下：
- 僅使用Spark SQL查詢
[root@cdh-master ~]# hadoop fs -ls oss://select-test-sz/data/lineitem.csv
-rw-rw-rw-1 2441079322 2018-10-31 11:18 oss://select-test-sz/data/lineitem.csv - 在Spark SQL上使用OSS Select查詢
scala> import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType, DoubleType}
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType, DoubleType}
scala> import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.{Row, SQLContext}
scala> val sqlContext = spark.sqlContext
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@74e2cfc5
scala> val textFile = sc.textFile("oss://select-test-sz/data/lineitem.csv")
textFile: org.apache.spark.rdd.RDD[String] = oss://select-test-sz/data/lineitem.csv MapPartitionsRDD[1] at textFile at <console>:26
scala> val dataRdd = textFile.map(_.split('|'))
dataRdd: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[2] at map at <console>:28
scala> val schema = StructType(
|List(
|StructField("L_ORDERKEY",LongType,true),
|StructField("L_PARTKEY",LongType,true),
|StructField("L_SUPPKEY",LongType,true),
|StructField("L_LINENUMBER",IntegerType,true),
|StructField("L_QUANTITY",DoubleType,true),
|StructField("L_EXTENDEDPRICE",DoubleType,true),
|StructField("L_DISCOUNT",DoubleType,true),
|StructField("L_TAX",DoubleType,true),
|StructField("L_RETURNFLAG",StringType,true),
|StructField("L_LINESTATUS",StringType,true),
|StructField("L_SHIPDATE",StringType,true),
|StructField("L_COMMITDATE",StringType,true),
|StructField("L_RECEIPTDATE",StringType,true),
|StructField("L_SHIPINSTRUCT",StringType,true),
|StructField("L_SHIPMODE",StringType,true),
|StructField("L_COMMENT",StringType,true)
|)
| )
schema: org.apache.spark.sql.types.StructType = StructType(StructField(L_ORDERKEY,LongType,true), StructField(L_PARTKEY,LongType,true), StructField(L_SUPPKEY,LongType,true), StructField(L_LINENUMBER,IntegerType,true), StructField(L_QUANTITY,DoubleType,true), StructField(L_EXTENDEDPRICE,DoubleType,true), StructField(L_DISCOUNT,DoubleType,true), StructField(L_TAX,DoubleType,true), StructField(L_RETURNFLAG,StringType,true), StructField(L_LINESTATUS,StringType,true), StructField(L_SHIPDATE,StringType,true), StructField(L_COMMITDATE,StringType,true), StructField(L_RECEIPTDATE,StringType,true), StructField(L_SHIPINSTRUCT,StringType,true), StructField(L_SHIPMODE,StringType,true), StructField(L_COMMENT,StringType,true))
scala> val dataRowRdd = dataRdd.map(p => Row(p(0).toLong, p(1).toLong, p(2).toLong, p(3).toInt, p(4).toDouble, p(5).toDouble, p(6).toDouble, p(7).toDouble, p(8), p(9), p(10), p(11), p(12), p(13), p(14), p(15)))
dataRowRdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[3] at map at <console>:30
scala> val dataFrame = sqlContext.createDataFrame(dataRowRdd, schema)
dataFrame: org.apache.spark.sql.DataFrame = [L_ORDERKEY: bigint, L_PARTKEY: bigint ... 14 more fields]
scala> dataFrame.createOrReplaceTempView("lineitem")
scala> spark.sql("select l_returnflag, l_linestatus, sum(l_quantity) as sum_qty, sum(l_extendedprice) as sum_base_price, sum(l_extendedprice * (1 - l_discount)) as sum_disc_price, sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge, avg(l_quantity) as avg_qty, avg(l_extendedprice) as avg_price, avg(l_discount) as avg_disc, count(*) as count_order from lineitem where l_shipdate > '1997-09-16' group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus").show()推薦閱讀
- 通過OSS 構(gòu)建大數(shù)據(jù)分析PB級(jí)數(shù)倉(cāng)
- 別克昂科威點(diǎn)煙器怎么使用
- 壁掛爐停暖如何使用 壁掛爐怎么停暖氣功能
- 《怪物獵人崛起》大劍流斬連段使用心得
- 使用洗衣機(jī)方式不當(dāng)比沒洗之前更臟
- 移動(dòng)式腳手架使用規(guī)范 移動(dòng)腳手架使用規(guī)范是什么
- 華為電話手表怎么使用
- 如何使用指南針確定方向,如何使用指南針確定方向的方法
- 寶駿510壽命
- 一輛汽車能開多少年
-
