spark sql percentile函数和spark dataframe分量器有什么不同？
Spark SQL percentile函数和Spark DataFrame分区器有以下几点不同之处：
- 功能不同：
- Spark SQL percentile函数是用于计算指定百分位数的函数。它可以用于计算数据集中某一列的百分位数，以便了解数据的分布情况。
- Spark DataFrame分区器是用于将数据集划分为多个分区的机制。它确定了数据在群集中的分布方式，以便在并行计算中提高性能。
- 使用方法不同：
- Spark SQL percentile函数是一个聚合函数，可在SQL查询中使用。您可以将其作为SELECT子句中的一部分使用，并指定要计算的百分位数。
- Spark DataFrame分区器是通过Spark编程API进行配置和使用的。您可以在创建DataFrame时指定分区的数量和分区键，或者使用repartition和coalesce等方法对现有的DataFrame进行重分区。
- 目的不同：
- Spark SQL percentile函数的主要目的是计算数据的分位数，以便进行数据分析和统计。
- Spark DataFrame分区器的主要目的是在分布式计算中提高性能和并行度。它可以帮助在集群中更好地分发和处理数据，以充分利用群集资源。
- 对应的腾讯云产品：
- 腾讯云的分析型数据库 ClickHouse 可以用于计算百分位数和其他统计指标，适用于大数据分析和数据仓库场景。详情请参考：ClickHouse 产品介绍
请注意，以上只是对两者之间一般性差异的描述，具体情况还需根据实际的使用场景和要求来确定。
相关·内容
我正在尝试计算DataFrame中列的百分位数？我在Spark聚合函数中找不到任何percentile_approx函数。
例如，在Hive中，我们有percentile_approx，我们可以按以下方式使用它
hiveContext.sql("select percentile_approx("Open_Rate",0.10) from myTable);
但出于性能原因，我想使用Spark DataFrame。
样本数据集
|User ID|Open_Rate|
-------------------
|A1 |10.3 |
|B1 |4
浏览 0提问于2016-06-07得票数 11
我使用的是spark 1.6。我需要为dataframe中的一列找到多个百分位数。我的数据非常庞大，至少有1000万条记录。我尝试使用配置单元上下文，如下所示
hivecontext.sql("select percentile_approx(col,0.25),percentile_approx(col,0.5) from table")
但是这种方法非常慢，而且需要大量的时间。我听说过approxQuantile，但似乎它在spark 2.x中可用。在spark 1.6中，是否有使用spark dataframe的替代方法来提高性能。
我看到了另一种使用hive UDAF的
浏览 14提问于2019-10-31得票数 0
回答已采纳
1回答
下面是Pandas代码，它返回输出
df_cols = df.columns.tolist()
spark_df_cols = spark_df['dic'].tolist()
i_names = []
i_quant = []
for i in df_cols:
if i in spark_df_cols:
i_names.append(i)
i_quant.append(df[i].quantile(0.9))
df_1 = pd.DataFrame({'dic': i_names, 'Pe
浏览 3提问于2022-06-14得票数 0
1回答
根据，百分位数命令应该给出每个数字列的确切百分位数。至少当我的输入是由浮点数组成时--这不是真的。
在中，它说百分位数命令只适用于整数。我不明白星星之火sql和hive之间的关系，但它似乎不是只运行hive --否则它就不会更改它的文档。另外，spark的percentile有一个不同的签名，允许它获得一个frequency参数，我也不知道它有什么用途。这是一个具有意外输出的示例：
from pyspark.sql import functions as sf
d = spark.createDataFrame([[35.138071000000004], [34.1199329999999
浏览 0提问于2019-07-21得票数 2
我已经写了一个pyspark函数，但当我运行多次时，它每次都给我同一组输入数据的不同输出。
-pyspark函数
def give_percentile(plat,metrics,perc):
df_perc = df_final.filter(df_final.platform.like('%' + plat + '%'))
df_perc = df_perc.filter(df_final[metrics]!=0)
percentile_val = df_perc.approxQuantile(metrics, [perc], 0.05)
if len(pe
浏览 0提问于2019-05-20得票数 0
我有一个Spark dataframe，它有Date，Group和Price列。
我正在尝试在Python语言中为该数据帧的Price列派生percentile(0.6)。此外，我需要将输出添加为新列。
我尝试了下面的代码：
perudf = udf(lambda x: x.quantile(.6))
df1 = df.withColumn("Percentile", df.groupBy("group").agg("group"),perudf('price'))
但它抛出了以下错误：
assert all(isinstance
浏览 0提问于2016-05-03得票数 4
软件包com.databricks:spark csv_2.11:1.2.0.使用SQLContext 1.导入org.apache.spark.sql.SQLContext 2. val sqlctx =新的SQLContext(sc) 3.导入sqlctx._
sqlctx.read.format("com.databricks.spark.csv").option("inferScheme"，df =.option(“定界符”，“；”).option(“头”，“真”).load(“/user/cloudera/data.csv”)
df.selec
浏览 15提问于2018-01-10得票数 0
我有Array[Row]，但实际上它总是只包含一个具有双值的Row。我想将这个双值提取到参数param1中。
val q1 = spark.sql("SELECT PERCENTILE(price,0.25) FROM table").collect()
val param1 = q1.mkString(",").toDouble
但我知道这个错误：
17/11/20 14:05:52 ERROR ApplicationMaster: User class threw exception: java.lang.NumberFormatException: Fo
浏览 1提问于2017-11-20得票数 1
回答已采纳
我正在使用谷歌商业应用程序接口，突然整个应用程序接口请求停止了，图表显示如下: REDUCE_PERCENTILE_99 你知道这是什么意思吗？ 谢谢。
浏览 11提问于2021-08-18得票数 0
我想把双倍分数转换成整数。我试过spark quantilediscretizer，但它太慢了。完成离散化过程需要几个小时。但是当我使用spark-sql的percentile函数时，它比quantilediscretizer快得多。那么，这两种方法之间有什么不同?在spark-sql中实现了哪些优化？
浏览 174提问于2019-06-24得票数 1
我正在创建一个hivecontext而不是sqlcontext来创建adtaframe
val conf=new SparkConf().setMaster("yarn-cluster")
val context=new SparkContext(conf)
//val sqlContext=new SQLContext(context)
val hiveContext=new HiveContext(context)
val data=Seq(1,2,3,4,5,6,7,8,9,10).map(x=>(x.toLong,x+1,x+2.toDouble)).toDF(&
浏览 0提问于2017-05-08得票数 0
我有一个非常大的CSV文件，它已经作为一个PySpark数据文件导入：df。dataframe包含许多列，包括列ireturn。我想要计算该列的0.99和0.01百分位数，然后将另一列添加到dataframe df中，作为new_col_99和new_col_01，它们分别包含0.99和0.01百分位数。我编写了下面的代码，它适用于小数据格式，但是当我将它应用到我的大型数据文件时会出现错误。
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv(&#
浏览 0提问于2019-01-15得票数 3
我得到以下错误：
18/03/14 15:31:11错误ApplicationMaster: User类抛出异常: org.apache.spark.sql.AnalysisException:表或视图未找到: products；第1行pos 42
这是我的密码：
val spark = SparkSession
.builder()
.appName("Test")
.getOrCreate()
val products = spark.read.parquet(prod
浏览 1提问于2018-03-14得票数 3
回答已采纳
Spark具有SQL函数percentile_approx()，其对应的Scala是df.stat.approxQuantile()。
但是，Scala对应项不能用于分组数据集，类似于df.groupby("foo").stat.approxQuantile()，如下所示：。
但是，可以在SQL语法中同时进行分组和百分位数。所以我想知道，也许我可以从SQL percentile_approx函数中定义一个UDF并在我的分组数据集中使用它？
浏览 2提问于2018-11-29得票数 8
回答已采纳
1回答
我想了解如何在Scala中实现相同方法。
def createDataFrame[A <: Product](data: Seq[A])(implicit evidence$3: reflect.runtime.universe.TypeTag[A]): org.apache.spark.sql.DataFrame
def createDataFrame(rdd: org.apache.spark.api.java.JavaRDD[_],beanClass: Class[_]): org.apache.spark.sql.DataFrame
def createDataFrame(row
浏览 0提问于2020-07-19得票数 0
回答已采纳
在spark.sql查询中注册和使用pyspark version 3.1.2内置函数的正确方式是什么？ 下面是一个创建pyspark DataFrame对象并在纯SQL中运行简单查询的最小示例。 尝试使用...TypeError: Invalid argument, not a string or column: -5 of type <class 'int'>. For column literals, use 'lit', 'array', 'struct' or 'create_map' fu
浏览 4提问于2021-08-19得票数 0
回答已采纳
我正在使用requests库对外部API进行HTTP调用。我的星火程序工作流是这样的：
(JSON_FILE:INPUT) ->(火花)->(HTTP)-> (KAFKA:OUTPUT)
当我运行它时，我会得到以下错误：
Caused by: java.io.NotSerializableException: requests.RequestAuth$Empty$
Serialization stack:
我遗漏了什么吗？
我的HTTP调用代码是：
import java.util.Properties
import org.apache.spark.sql.funct
浏览 1提问于2021-02-02得票数 1
回答已采纳
1回答
我运行pyspark2在spark 2.0.0上
考虑到加载到Dataframe中的稳定和常量数据集，我确实按特定列删除了重复数据：
vw_ticket = read_csv(...)
vw_ticket_in_booking = vw_ticket.dropDuplicates(['BOOKING_TICKET_ID'])
vw_ticket_in_booking.createOrReplaceTempView('vw_ticket_in_booking')
然后
spark.sql('select count(distinct(booking_id)
浏览 2提问于2017-08-31得票数 0
回答已采纳
我有两个数据文件，如下所示。
dataframe1 =
Index features constant
0 AA 0.25
1 AB 0.45
2 AC 0.78
3 AD 0.91
4 AE 0.12
dataframe2 =
Index AA AB AC AD AE
0 10 45 15 14 98
1 14 55 55 48 70
2 25 75
浏览 4提问于2020-01-27得票数 0
回答已采纳
我是新来的火花环境。我有列名如下的数据集：
user_id，Date_time，order_quantity
我要计算每一个order_quantity的第90个百分位数。
如果是sql，我将使用以下查询：
%sql
SELECT user_id, PERCENTILE_CONT ( 0.9 ) WITHIN GROUP (ORDER BY order_quantity) OVER (PARTITION BY user_id)
但是，spark没有内置的支持使用percentile_cont函数的支持。
对于我如何在上面的数据集中实现这一点，有什么建议吗？如果需要更多的信息，请告诉我。
浏览 1提问于2016-11-10得票数 4
1回答
我试图在spark中使用百分位数函数。
数据：
col1
----
198
15.8
198
198
198
198
198
198
198
198
198
如果我使用下面的代码，我得到的百分位数是不正确的。
select percentile('col1', .05) from tblname
产出: 106.9
如果我使用下面的代码，我得到的百分位数是不正确的。
select percentile('col1', .05, 2) from tblname
产出: 24.91000000000001
但是如果我使用下面的代码，我就会得到预期的答
浏览 0提问于2019-05-27得票数 4
我试图将一个列的数据main添加到更大的dataframe中，但是，第一个dataframe的问题是，在创建它之后，尝试通过命令将其添加到主dataframe中：
df.withColumn("name", dataframe)
我知道错误：
**found : org.apache.spark.sql.DataFrame
(which expands to) org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
required: org.apache.spark.sql.Column**
我知道Data
浏览 0提问于2018-05-20得票数 3
回答已采纳
我已经给出了示例表。我想从"value“列获得每组"source”列的中位数。其中，源列是字符串，DataType值列是双DataType
scala> sqlContext.sql("SELECT * from tTab order by source").show
+---------------+-----+
| Source|value|
+---------------+-----+
|131.183.222.110
浏览 10提问于2015-12-30得票数 7
我试着创建Apache Spark dataframe
val valuesCol = Seq(("Male","2019-09-06"),("Female","2019-09-06"),("Male","2019-09-07"))
valuesCol: Seq[(String, String)] = List((Male,2019-09-06), (Female,2019-09-06), (Male,2019-09-07))
模式
val someSchema = List(StructField
浏览 8提问于2019-09-19得票数 0
回答已采纳
2回答
在SQL中计算百分位
、、、
这应该是非常简单的，但作为SQL的新手，我真的很挣扎。有人建议我对连续(非离散)数据使用PERCENTILE_CONT。
所讨论的数据涉及两列：(1)患者列表的in和(2)每年的平均事件数。
使用我在网上找到的一些代码，这就是我要做的
SELECT ID,
percentile_cont (0.25) WITHIN GROUP
(ORDER BY PPPY ASC) OVER(PARTITION BY ID) as percentile_25,
percentile_cont (0.50) WITHIN GROUP
(ORDER BY PPPY ASC) OVER(PARTITION BY I
浏览 20提问于2018-07-26得票数 5
当我试图将输入表作为视图查询时，我会得到错误com.palantir.foundry.spark.api.errors.DatasetPathNotFoundException。我的代码如下：
def Median_Product_Revenue_Temp2(Merchant_Segments):
Merchant_Segments.createOrReplaceTempView('Merchant_Segments_View')
df = spark.sql('select * from Merchant_Segments_View limit 5&
浏览 13提问于2022-02-25得票数 3
import numpy as np
import pandas as pd
import math
from scipy import stats
hqm_columns = [
'Ticker',
'Price',
'Number of Shares to Buy',
'One-Year Price Return',
'One-Year Return Percentile',
'Six-Month Price Return',
浏览 24提问于2021-02-18得票数 1
我是一个新的火花。
谢谢你的关注和帮助。这是我的问题。我在mac os （Catalina 10.15.7）中以本地模式启动了我的spark-shell（version 3.2.1)，然后运行了下面的代码，没有出现异常。
import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}
import org.apache.spark.sql.Row
import org.apache.spark.sql.DataFrame
import org.apache.spark.rdd.RDD
v
浏览 3提问于2022-06-22得票数 0
回答已采纳
3回答
我有一个字符串，其中的函数需要进入我预期的dataframe的.agg函数。我的数据数据看起来像
val client = Seq((1,"A","D",10),(2,"A","D",5),(3,"B","C",56),(5,"B","D",67)).toDF("ID","Categ","subCat","Amnt")
+---+-----+------+----+
| ID|Categ|subCat|A
浏览 1提问于2020-01-19得票数 1
2回答
系统: Spark 1.3.0 (Anaconda Python dist.)在Cloudera Quickstart VM 5.4上
这是一个Spark DataFrame：
from pyspark.sql import SQLContext
from pyspark.sql.types import *
sqlContext = SQLContext(sc)
data = sc.parallelize([('Foo',41,'US',3),
('Foo',39,'UK',1),
浏览 3提问于2015-05-15得票数 3
回答已采纳
1回答
我是新来的火花/斯卡拉。我正在尝试读取一些数据从一个蜂窝表到一个火花数据，然后添加一个列的基础上的一些条件。这是我的代码：
val DF = hiveContext.sql("select * from (select * from test_table where partition_date='2017-11-22') a JOIN (select max(id) as bid from test_table where partition_date='2017-11-22' group by at_id) b ON a.id=b.bid")
浏览 1提问于2017-11-27得票数 0
回答已采纳
1回答
我试图通过在函数中持久化数据帧来重用它。
def run() {
val df1 = getdf1(spark: SparkSession)
val df2 = getdf2(spark:SparkSession, df1)
}
def getdf1(spark: SparkSession): DataFrame {
val sqltxt = "select * from emp"
val df1 = spark.sql(sqltxt)
df1.persist
spark.sql("SET spark.sql.hive.convertMetas
浏览 2提问于2018-03-09得票数 0
有时(例如，为了测试和标记)，我想强制执行在DataFrame上定义的转换。AFAIK调用像count这样的操作并不能确保所有的Columns都是实际计算的，show可能只计算所有Rows的一个子集(参见下面的示例) 我的解决方案是使用df.write.saveAsTable将DataFrame写到HDFS，但是这会“扰乱”我的系统，我不想再保存更多的表。 那么触发DataFrame求值的最佳方式是什么呢 编辑： 请注意，在spark开发人员列表上还有一个最近的讨论：http://apache-spark-developers-list.1001551.n3.nabble.com/Will-
浏览 62提问于2017-03-10得票数 20
回答已采纳
Spark SQL中Group By子句的底层实现是什么？我知道Spark支持下面两种类型的Group by操作，即GroupByKey和ReduceByKey。ReduceByKey是一种map side reduce，它提供了比GroupByKey更好的性能。
在我们的应用程序代码中，我们在Spark Dataframe上使用Spark SQL，而不是直接创建RDDs。所以，我想到了这个问题，Spark SQL中的GroupBy是做GroupByKey还是ReduceByKey，还是别的什么。
浏览 0提问于2019-08-30得票数 1
我使用星火1.6.1。在我的Spark程序中，我连接到Postgres数据库，并通过JDBC将每个表注册为临时表。例如：
Map<String, String> optionsTable = new HashMap<String, String>();
optionsTable.put("url", "jdbc:postgresql://localhost/database?user=postgres&password=passwd");
optionsTable.put("dbtable", "table
浏览 0提问于2016-04-16得票数 3
回答已采纳
我正在尝试将json文件序列化为parquet格式。我有一个错误：
org.apache.spark.api.java.function.VoidFunction2[org.apache.spark.sql.Datasetorg.apache.spark.sql.Row，java.lang.Long])org.apache.spark.sql.streaming.DataStreamWriterorg.apache.spark.sql.streaming.DataStreamWriterorg.apache.spark.sql.Row错误：(34，25)重载方法foreachBatch与备选方
浏览 5提问于2020-07-28得票数 6
1回答
我已经在基于SP3的SLES 11的HDP2.3上安装了Vora1.2。
当我试图按照新安装和管理手册第2.7节(第34页)对Vora进行基于命令行的验证时，我现在得到了一个新的错误：
scala> vc.sql(testsql)
com.sap.spark.vora.discovery.DiscoveryException: Could not connect to Consul Agent on localhost:8500 : null at
com.sap.spark.vora.discovery.ConsulDiscoveryClient$ConsulDisc
浏览 0提问于2016-04-05得票数 0
回答已采纳
3回答
我正在尝试过滤RDD，如下所示：
spark_df = sc.createDataFrame(pandas_df)
spark_df.filter(lambda r: str(r['target']).startswith('good'))
spark_df.take(5)
但是得到了以下错误：
TypeErrorTraceback (most recent call last)
<ipython-input-8-86cfb363dd8b> in <module>()
1 spark_df = sc.createDataFra
浏览 2提问于2016-10-06得票数 18
我正在尝试从数据列表创建一个DataFrame，并希望在其上应用模式。在Spark Scala文档中，我尝试使用这个接受行列表和模式作为StructType的createDataframe签名。 def createDataFrame(rows: List[Row], schema: StructType): DataFrame 下面是我正在尝试的示例代码 import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
val simpleData = List(Row("James", "Sal
浏览 36提问于2020-10-01得票数 0
回答已采纳
2回答
下面的代码可以正常工作，直到我在show之后添加agg。为什么show是不可能的？
val tempTableB = tableB.groupBy("idB")
.agg(first("numB").as("numB")) //when I add a .show here, it doesn't work
tableA.join(tempTableB, $"idA" === $"idB", "inner")
.drop("idA", "numA"
浏览 0提问于2017-07-26得票数 2
回答已采纳
我使用的是Spark 3.0中的结构化流媒体。 我正在尝试做的是将数据写入多个接收器。我需要写一些卡夫卡在另一个进程中使用的DataFrame，也需要存储在卡桑德拉相同的一个供以后使用(一些仪表板等)。 对于目标进程，我编写了如下代码。我参考了here的官方文档。 merged_stream.writeStream
//.trigger(Trigger.ProcessingTime("3 seconds"))
.foreachBatch((batchDF: DataFrame, batchId: Long) => {
batch
浏览 59提问于2020-08-29得票数 3
回答已采纳
我试图使用星火红移库，并且无法操作由sqlContext.read()命令创建的数据(从redshift读取)。
这是我的代码：
Class.forName("com.amazon.redshift.jdbc41.Driver")
val conf = new SparkConf().setAppName("Spark Application").setMaster("local[2]")
val sc = new SparkContext(conf)
import org.apache.spark.sql._
val sqlContext
浏览 4提问于2015-11-04得票数 1
回答已采纳
2回答
我正在尝试使用窗口函数对列进行百分位数计算，如下所示。我已经让在组上使用ApproxQuantile定义。
val df1 = Seq(
(1, 10.0), (1, 20.0), (1, 40.6), (1, 15.6), (1, 17.6), (1, 25.6),
(1, 39.6), (2, 20.5), (2 ,70.3), (2, 69.4), (2, 74.4), (2, 45.4),
(3, 60.6), (3, 80.6), (4, 30.6), (4, 90.6)
).toDF("ID","Count")
val id
浏览 35提问于2020-06-05得票数 2
回答已采纳
实际上，我了解是如何工作的，但是通过手工计算百分位数和调用函数，我得到了不同的结果。以下是数据集：
305.7884804
350
373.3728865
384.2094838
410.8603441
414.9842786
455.3545205
550
计算25%时，我将350和373.3728865之和除以2，得到的结果是361.68644325。
计算50%时，我将384.2094838和410.8603441之和除以2，得到的结果是397.53491395。
计算75%时，我将414.9842786和455.3545205之和除以2，得到的结果是435.16939955。
但是，当
浏览 4提问于2016-06-06得票数 4
回答已采纳
我想在Palantir Foundry代码库中定义以下pandas_udf。
@pandas_udf("long", PandasUDFType.GROUPED_AGG)
def percentile_95_udf(v):
return v.quantile(0.95)
但是，当我试图在全局范围内定义这个udf时，我会得到以下错误：
AttributeError: 'NoneType' object has no attribute '_jvm'
但是，如果我在转换调用的函数中定义了相同的函数，则代码运行良好，如下所示：
from pys
浏览 2提问于2021-03-02得票数 1
回答已采纳
1回答
我学的是火种。例如，我正在尝试从sql构建DataFrame
DF=spark.sql("with a as (select ....) select ...")
我的sql有点复杂，所以要执行20分钟。我觉得DF是对我的SQL的引用，它意味着当我执行DF.head(10)需要20分钟，下一步DF.count()也需要20分钟等等。
我想让DataFrame像熊猫一样在内存中有价值，其中DF.head(10)，DF.count()需要几秒钟。
我唯一能想到的方法是使用"create“，例如：
xx=spark.sql("create table yyy as w
浏览 3提问于2022-04-04得票数 0
spark sql如何使用Dataframe读取apache cassandra数据ttl (Spark Sql)。我找不到任何例子？我能举个例子吗？
我之前的问题是关于rdd Does：
但是现在查询是针对Dataframe的。
浏览 23提问于2020-02-10得票数 0
我有以下数据帧 ------------------------------------------------------
|longitude | latitude | geomType | geom |
------------------------------------------------------
|-7.0737816 |33.82666166|Polygon |[GEOMETRY - 113 o]|
------------------------------------------------------- 我想将此查询应用于
浏览 55提问于2020-07-09得票数 0
我有一个cassandraSQLContext，在这里我可以做到这一点：
cassandraSqlContext.setKeyspace("test");
因为如果我不这样做，它会抱怨我设置了默认的键空间。
现在当我运行这段代码时：
def insertIntoCassandra(siteMetaData: MetaData, dataFrame: DataFrame): Unit ={
System.out.println(dataFrame.show())
val tableName = siteMetaData.getTable
浏览 0提问于2017-02-28得票数 0
1回答
我正在尝试创建一个具有动态模式生成的Dataframe。下面是代码片段：
def mapMetricList(row: Row): Seq[Metric] = ???
val fields = Seq("Field1", "Field2")
case class Metric(name: String, count: Long)
def convertMetricList(df: DataFrame): DataFrame = {
val outputFields = df.schema.fieldNames.filter(f => fields.
浏览 0提问于2017-09-05得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
