0

我是 Apache Spark 和 Scala 的新手(通常也是 Hadoop 的初学者)。我完成了 Spark SQL 教程:https ://spark.apache.org/docs/latest/sql-programming-guide.html 我尝试对标准 csv 文件执行简单查询,以在当前集群上对其性能进行基准测试。

我使用来自https://s3.amazonaws.com/hw-sandbox/tutorial1/NYSE-2000-2001.tsv.gz的数据,将其转换为 csv 并复制/粘贴数据以使其大 10 倍。

我使用 Scala 将它加载到 Spark 中:

// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD.
import sqlContext.createSchemaRDD

定义类:

case class datum(exchange: String,stock_symbol: String,date: String,stock_price_open: Double,stock_price_high: Double,stock_price_low: Double,stock_price_close: Double,stock_volume: String,stock_price_adj_close: Double)

读入数据:

val data = sc.textFile("input.csv").map(_.split(";")).filter(line => "exchange" != "exchange").map(p => datum(p(0).trim.toString, p(1).trim.toString, p(2).trim.toString, p(3).trim.toDouble, p(4).trim.toDouble, p(5).trim.toDouble, p(6).trim.toDouble, p(7).trim.toString, p(8).trim.toDouble))

转换为表格:

data.registerAsTable("data")

定义查询(列出所有以“IBM”为股票代码的行):

val IBMs = sqlContext.sql("SELECT * FROM data WHERE stock_symbol ='IBM'")

执行计数以便查询实际运行:

IBMs.count()

查询运行良好,但返回 res: 0 而不是 5000(这是使用 Hive 和 MapReduce 返回的结果)。

4

1 回答 1

5

filter(line => "exchange" != "exchange")

由于“exchange”等于“exchange”,过滤器将返回一个大小为 0 的集合。由于没有数据,因此查询任何结果都会返回 0。您需要重新编写逻辑。

于 2014-09-08T15:57:12.307 回答