如何使用 Spark 中的组聚合滚动时间窗口
我有一些数据要按特定列分组，然后根据组中的滚动时间窗口聚合一系列字段. 以下是一些示例数据: df = spark.createDataFrame([Row(date='2016-01-01', group_by='group1', get_avg=5, get_first=1),行(日期='2016-01-10', group_by='group1', get_avg=5, get_fi
..
apache-spark-sql相关内容
PySpark - 使用 withColumnRenamed 重命名多列
我想使用 spark withColumnRenamed 函数更改两列的名称.当然，我可以写: data = sqlContext.createDataFrame([(1,2), (3,4)], ['x1', 'x2'])数据 = (数据.withColumnRenamed('x1','x3').withColumnRenamed('x2', 'x4')) 但我想一步完成(有新名称的列表/元组
..
按日期从 Spark 中的 S3 读取多个文件
说明 我有一个应用程序，它将数据发送到 AWS Kinesis Firehose 并将数据写入我的 S3 存储桶.Firehose 使用“yyyy/MM/dd/HH"格式写入文件. 就像这个示例 S3 路径: s3://mybucket/2016/07/29/12 现在我有一个用 Scala 编写的 Spark 应用程序，我需要在其中读取特定时间段的数据.我有开始和结束日期.数据采
..
选择数据时 Hive 是否保留文件顺序
如果我这样做 select * from table1; 检索数据的顺序 文件顺序或随机顺序 解决方案 如果没有 ORDER BY，则无法保证顺序. 许多进程(映射器)并行读取数据，在计算拆分后，每个进程开始读取一些文件或几个文件，具体取决于计算的拆分. 所有的并行进程可以处理不同的数据量并运行在不同的节点上，每次的负载都不一样，所以它们开始返回行并在不同的时间完成，这取
..
YARN 中 Spark 应用程序的物理内存不断增加
我在 YARN 中运行一个 Spark 应用程序，它有两个执行程序，Xms/Xmx 为 32GB，spark.yarn.excutor.memoryOverhead 为 6GB. 我看到应用程序的物理内存不断增加，最终被节点管理器杀死: 2015-07-25 15:07:05,354 WARN org.apache.hadoop.yarn.server.nodemanager.contai
..
Spark sql 每组前 n 个
如何在 spark-sql 中获得每个组的前 n(比如前 10 名或前 3 名)? http://www.xaprb.com/blog/2006/12/07/how-to-select-the-firstleastmax-row-per-group-in-sql/ 提供了一般教程SQL.但是，spark 并没有在 where 子句中实现子查询. 解决方案 可以使用 Spark 1.4
..
从列表 PySpark 创建单行数据框
我有这样的数据 data = [[1.1, 1.2], [1.3, 1.4], [1.5, 1.6]] 我想创建一个 PySpark 数据框 我已经用了 dataframe = SQLContext.createDataFrame(data, ['features']) 但我总是得到 +--------+---+|特点|_2|+--------+---+|1.1|1.2||1.3|1.
..
GroupBy 列并过滤 Pyspark 中具有最大值的行
我几乎可以肯定以前有人问过这个问题，但是
..
如何检查火花数据框是否为空?
现在，我必须使用 df.count >0 检查 DataFrame 是否为空.但这有点低效.有没有更好的方法来做到这一点? 谢谢. PS:我想检查它是否为空，以便我只保存 DataFrame 如果它不为空 解决方案 对于 Spark 2.1.0，我的建议是使用 head(n: Int) 或 take(n:Int) 和 isEmpty，以您最清楚的意图为准. df.head(1
..
Spark SQL - 使用 JDBC 加载数据，使用 SQL 语句，而不是表名
我想我遗漏了一些东西，但不知道是什么.我想使用 SQLContext 和 JDBC 使用特定的 sql 语句加载数据喜欢 使用(nolock)从table1中选择前1000个文本其中 threadid 在 (使用 (nolock) 从 table2 中选择不同的 id其中 flag=2 and date >='1/1/2015' and userid in (1, 2, 3)) 我应该使用哪种
..
Spark/Scala:前向填充最后一次观察
使用 Spark 1.4.0、Scala 2.10 我一直在想办法用最后一次已知的观察结果来向前填充空值，但我没有看到简单的方法.我认为这是很常见的事情，但找不到说明如何执行此操作的示例. 我看到了用一个值向前填充 NaN 的函数，或者用一个偏移量填充或移动数据的滞后/超前函数，但没有找到最后一个已知值. 在网上看，我在 R 中看到很多关于同一件事的问答，但在 Spark/Sca
..
Spark Dataframe :How to add a index Column : Aka Distributed Data Index
我从 csv 文件中读取数据，但没有索引. 我想从 1 添加一列到行号. 我该怎么办，谢谢(scala) 解决方案 使用 Scala，您可以使用: import org.apache.spark.sql.functions._df.withColumn("id",monotonicallyIncreasingId) 你可以参考这个示例 和 Scala 文档. 使用 P
..
Spark 从一行中提取值
我有以下数据框 val transactions_with_counts = sqlContext.sql("""SELECT user_id AS user_id, category_id AS category_id,COUNT(category_id) FROM 交易 GROUP BY user_id, category_id""") 我正在尝试将行转换为 Rating 对象，但由于 x
..
对火花数据帧的同一列进行多个聚合操作
我有三个包含以下信息的字符串类型数组: groupBy 数组:包含我想要对数据进行分组的列的名称. 聚合数组:包含我要聚合的列的名称. operations 数组:包含我想要执行的聚合操作 我正在尝试使用 spark 数据框来实现这一点.Spark 数据帧提供了一个 agg()，您可以在其中传递 Map [String,String](列名和各自的聚合操作)作为输入，但是我想对数据
..
如何使用 Hive 支持创建 SparkSession(因“找不到 Hive 类"而失败)?
我在尝试运行以下代码时遇到错误: import org.apache.spark.sql.Dataset;导入 org.apache.spark.sql.Row;导入 org.apache.spark.sql.SparkSession;公共类应用{public static void main(String[] args) 抛出异常 {火花会话.builder().enableHiveSuppo
..
如何展平 Spark 数据帧中的结构?
我有一个具有以下结构的数据框: |-- data: struct (nullable = true)||-- id: long (nullable = true)||-- keyNote: struct (nullable = true)|||-- 键:字符串(可为空 = 真)|||-- 注意:字符串(可为空 = 真)||-- 详细信息:地图(可为空 = 真)|||-- 键:字符串|||--
..
选择数组中的一系列元素spark sql
我使用 spark-shell 进行以下操作. 最近在 spark-sql 中加载了一个包含数组列的表. 这是相同的 DDL: 创建表 test_emp_arr{dept_id 字符串，dept_nm 字符串，emp_details 数组} 数据看起来像这样 +-------+-------+----------------------------------+|dept_i
..
从 Spark SQL 中的字符串列表创建文字和列数组
我试图在 Scala 中定义函数，将字符串列表作为输入，并将它们转换为传递给下面代码中使用的数据帧数组参数的列. val df = sc.parallelize(Array((1,1),(2,2),(3,3))).toDF("foo","bar")val df2 = df.withColumn("columnArray",array(df("foo").cast("String"),df("ba
..
在 Spark 中访问数组列
Spark DataFrame 包含一个 Array[Double] 类型的列.当我尝试在 map() 函数中取回它时，它会抛出 ClassCastException 异常.以下 Scala 代码生成异常. case class Dummy( x:Array[Double] )val df = sqlContext.createDataFrame(Seq(Dummy(Array(1,2,3)))
..
Spark 错误:用于构建 ClassDict 的预期零参数(对于 numpy.core.multiarray._reconstruct)
我在 Spark 中有一个数据框，其中一列包含一个数组.现在，我编写了一个单独的 UDF，它将数组转换为另一个仅包含不同值的数组.请参见下面的示例: 例如:[24,23,27,23] 应该转换为 [24,23,27]代码: def uniq_array(col_array):x = np.unique(col_array)返回 xuniq_array_udf = udf(uniq_arra
..