标签: apache-spark-sql

我想用Scala API显示整个Apache Spark SQL DataFrame.我可以使用这个show()方法:

myDataFrame.show(Int.MaxValue)

有没有比使用更好的方式来显示整个DataFrame Int.MaxValue？

pyspark中有一个DataFrame,数据如下:

user_id object_id score
user_1  object_1  3
user_1  object_1  1
user_1  object_2  2
user_2  object_1  5
user_2  object_2  2
user_2  object_2  6

我期望在每个组中返回具有相同user_id的2条记录,这些记录需要具有最高分.因此,结果应如下所示:

user_id object_id score
user_1  object_1  3
user_1  object_2  2
user_2  object_2  6
user_2  object_1  5

我是pyspark的新手,有人能给我一个代码片段或门户网站来解决这个问题的相关文档吗？十分感谢!

我有这个代码:

l = [('Alice', 1),('Jim',2),('Sandra',3)]
df = sqlContext.createDataFrame(l, ['name', 'age'])
df.withColumn('age2', df.age + 2).toPandas()

工作正常,做它需要的东西.假设我只想显示前n行,然后调用toPandas()返回pandas数据帧.我该怎么做？我无法调用,take(n)因为它不会返回数据帧,因此我无法将其传递给toPandas().

换句话说,如何从数据帧中获取前n行并调用toPandas()结果数据帧？想不到这很难但我无法弄清楚.

我正在使用Spark 1.6.0.

我查看了文档,并说它支持以下连接类型:

要执行的联接类型.默认内心.必须是以下之一:inner,cross,outer,full,full_outer,left,left_outer,right,right_outer,left_semi,left_anti.

我查看了关于SQL连接的StackOverflow答案,并且顶部几个答案没有提到上面的一些连接,例如left_semi和left_anti.他们在Spark中意味着什么？

有没有办法RDD在spark中连接两个不同s的数据集？

要求是 - 我使用具有相同列名的scala创建两个中间RDD,需要组合这两个RDD的结果并缓存访问UI的结果.如何在此处组合数据集？

RDD属于类型 spark.sql.SchemaRDD

Spark 1.4.1

我遇到一种情况,按数据框进行分组,然后对'count'列进行计数和过滤会引发下面的异常

import sqlContext.implicits._
import org.apache.spark.sql._

case class Paf(x:Int)
val myData = Seq(Paf(2), Paf(1), Paf(2))
val df = sc.parallelize(myData, 2).toDF()

然后分组和过滤:

df.groupBy("x").count()
  .filter("count >= 2")
  .show()

引发异常:

java.lang.RuntimeException: [1.7] failure: ``('' expected but `>=' found count >= 2

解:

重命名列会使问题消失(因为我怀疑与插值'count'函数没有冲突'

df.groupBy("x").count()
  .withColumnRenamed("count", "n")
  .filter("n >= 2")
  .show()

那么,这是一种期望的行为,一个错误还是一种规范的方式？

谢谢,亚历克斯

我想DataFrame使用与列长度相关的条件来过滤a ,这个问题可能很容易,但我没有在SO中找到任何相关的问题.

更具体的,我有一个DataFrame只有一个Column,其中ArrayType(StringType()),我要筛选的DataFrame使用长度filterer,我拍下面的一个片段.

df = sqlContext.read.parquet("letters.parquet")
df.show()

# The output will be 
# +------------+
# |      tokens|
# +------------+
# |[L, S, Y, S]|
# |[L, V, I, S]|
# |[I, A, N, A]|
# |[I, L, S, A]|
# |[E, N, N, Y]|
# |[E, I, M, A]|
# |[O, A, N, A]|
# |   [S, U, S]|
# +------------+

# But I want only the entries with length …

引用Spark DataFrames,Datasets和SQL手册:

Spark中还没有包含少数Hive优化.由于Spark SQL的内存计算模型,其中一些(如索引)不太重要.其他版本用于将来的Spark SQL版本.

作为Spark的新手,我有点困惑,原因有两个:

1. Spark SQL旨在处理大数据,至少在我的用例中,数据大小远远超过可用内存的大小.假设这并不罕见,"Spark SQL的内存计算模型"是什么意思？Spark SQL是否仅适用于数据适合内存的情况？

2. 即使假设数据适合内存,对非常大的数据集进行全扫描也需要很长时间.我在内存数据库中读取了这个反对索引的论点,但我并不相信.那里的例子讨论了一个10,000,000记录表的扫描,但这不是真正的大数据.扫描具有数十亿条记录的表可能导致"SELECT x WHERE y = z"类型的简单查询永远不会立即返回.

我理解索引有诸如较慢的INSERT/UPDATE,空间要求等缺点.但在我的用例中,我首先处理并将大批数据加载到Spark SQL中,然后将这些数据作为一个整体进行探索,而无需进一步修改.Spark SQL对于初始分布式处理和数据加载很有用,但缺乏索引使得交互式探索比我预期的更慢,更麻烦.

我想知道为什么Spark SQL团队认为索引不重要到它的路线图之外的程度.是否有不同的使用模式可以提供索引的好处,而无需独立实现等效的东西？

我试图将csv文件读入数据帧.我知道我的数据帧的架构应该是什么,因为我知道我的csv文件.另外我使用spark csv包来读取文件.我试图指定如下的架构.

val pagecount = sqlContext.read.format("csv")
  .option("delimiter"," ").option("quote","")
  .option("schema","project: string ,article: string ,requests: integer ,bytes_served: long")
  .load("dbfs:/databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample/pagecounts-20151124-170000")

但是当我检查我创建的数据框架的模式时,它似乎采用了自己的模式.我做错了吗？如何制作火花来接收我提到的架构？

> pagecount.printSchema
root
|-- _c0: string (nullable = true)
|-- _c1: string (nullable = true)
|-- _c2: string (nullable = true)
|-- _c3: string (nullable = true)

我是Spark和Spark SQL的新手.

如何createOrReplaceTempView在Spark工作？

如果我们将一个RDD对象注册为一个表,那么火花会将所有数据保存在内存中吗？