当连接键以列表形式给出时，如何修改spark dataframe中连接的列？
在Spark DataFrame中，可以使用withColumnRenamed方法来修改连接的列。该方法接受两个参数，第一个参数是要修改的列名，第二个参数是修改后的列名。
以下是修改连接列的示例代码：
from pyspark.sql import SparkSession
# 创建SparkSession
spark = SparkSession.builder.getOrCreate()
# 创建示例DataFrame
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, ["name", "age"])
# 修改连接列
df = df.withColumnRenamed("name", "new_name")
# 显示修改后的DataFrame
df.show()输出结果为：
+---------+---+
|new_name |age|
+---------+---+
|Alice |25 |
|Bob |30 |
|Charlie |35 |
+---------+---+在上述示例中，我们使用withColumnRenamed方法将原始的"name"列修改为"new_name"列。
相关·内容
1回答
我试图使用以下代码从cosmos db中读取30K行数据，使用spark连接器
val readConfig = Config(Map(
"Endpoint" -> "",
"Masterkey" -> "",
"Database" -> "",
"Collection" -> "",
"PreferredRegions" -> "",
"query_custom"
浏览 0提问于2021-05-22得票数 3
回答已采纳
3回答
我正在尝试使用SparkSQL对数据帧执行广播散列联接，如下所示：
在该示例中，(小) DataFrame通过saveAsTable持久化，然后通过spark SQL (即通过sqlContext.sql("...")) )进行连接
我遇到的问题是，我需要使用SQL来构造我的sparkSQL (我需要用一个ID列表连接大约50个表，并且不想手动编写这个SQL语句)。
How do I tell spark to use the broadcast hash join via the API? The issue is that if I load the ID list (f
浏览 7提问于2016-05-27得票数 16
回答已采纳
1回答
我想知道有没有办法把它自动化。我想做一个函数，在这个函数中，我会告诉你，我想要加入多少列。如果我有3列的dataFrame，并给出一个参数"number_of_columns=3"，那么它将连接列: 0，1，2。但如果我有7列的dataFrame，并给出参数"number_of_columns=7"，那么它将连接列: 0，1，2，3，4，5，6。列的名称总是相同的:从"0“到"number_ of _ columns -1”。 有没有办法做到这一点？或者，如果我有其他数量的列要合并，我必须有另一个函数？ def my_function(spar
浏览 12提问于2021-04-01得票数 2
回答已采纳
我有一个Seq of Spark (即Seq[org.apache.spark.sql.DataFrame])，它可以包含一个或多个元素。有一个列的列表对每个数据文件都是通用的，每个dataframe也有一些额外的列。我想要做的是使用连接条件中的那些公共列将所有这些数据文件连接在一起(记住，数据格式的数量是未知的)
我怎样才能把所有这些数据文件连接起来？我想我可以对它们进行foreach，但这似乎不太优雅。谁能想出一种更实用的方法来做这件事呢？编辑:递归函数会比预测更好，我现在正在做这个工作，完成后会在这里发布。
下面是一些代码，它创建了n个dataframes列表(在本例中是n=3)，每个列
浏览 0提问于2018-05-15得票数 4
回答已采纳
我正在尝试向我的Spark Dataframe添加一个新列。添加的新列的大小将基于变量(例如salt) post，我将使用该列分解数据集以用于盐连接。 目前，我在array函数中使用连续的lit，但这有一个问题，它不能被参数化，而且作为一种编码实践看起来很糟糕。我目前的实现看起来像下面这样。 int salt =3;
Dataset<Row> Reference_with_Salt_Col = Reference.withColumn("salt_array", array(lit(0), lit(1), lit(2))); 我参考和研究了各种方法，但似乎都不能
浏览 80提问于2021-03-14得票数 1
回答已采纳
我希望根据一个日期时间列(格式为"23-07-2018 16:01" )，按升序重新排序数据。
我的程序排序到日期级别，但不是 HH:mm 标准，我希望输出包含HH:mm细节，并根据它进行排序。
package com.spark
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{to_date, to_timestamp}
import org.apache.spark.sql.functio
浏览 0提问于2018-08-07得票数 0
我试图使用Pyspark在dataframe中更改一个列，也就是跨行对单个列的所有值进行洗牌。
我试图避免这样的解决方案，即在将列拆分并分配索引列之前，将其重新连接到原始的dataframe，而原始dataframe也有一个添加的索引列。主要是因为我的理解(这可能是非常错误的)，在大型数据集(数百万行)的运行时，联接是不好的。
# for some dataframe spark_df
new_df = spark_df.select(colname).sort(colname)
new_df.show() # column values sorted nicely
spark_df.with
浏览 0提问于2019-06-06得票数 0
我有一个下面的Cassandra表模式。 ColumnA Primary Key
ColumnB Clustering Key
ColumnC
ColumnD 现在，我有了一个Spark RDD，它的列排序为RDDColumnC、ColumnA、ColumnB、ColumnD 因此，当我写入Cassandra表时，我需要确保顺序是正确的。因此，我使用SomeColumns指定了列的排序 rdd.saveToCassandra(keyspace,table,SomeColumns("ColumnA","ColumnB","ColumnC",
浏览 10提问于2019-07-02得票数 0
回答已采纳
3回答
是否有任何方法将新列附加到现有的拼花文件中？
我目前正在进行一场kaggle竞赛，我已经把所有的数据转换成了拼花文件。
下面是这样的情况，我将拼花文件读入pyspark中，进行了一些特性提取，并将新列添加到DataFrame中
pysaprk.DataFrame.withColumn()。
之后，我想将新列保存在源文件中。
我知道Spark随而来，但是这个例子只给出了一个键值的情况。
拼花“附加”模式也做不到这一点。它只将新行附加到拼花文件中。如果要将新列追加到现有的parquet文件中，而不是再次生成整个表，那么呢？或者，我必须生成一个单独的新拼板文件并在运行时将它们连接起来。
浏览 1提问于2015-08-04得票数 14
基于前面的问题：，。假设我有以下数据：
df = spark.createDataFrame(
[(1, "a", 23.0), (3, "B", -23.0)],
("x1", "x2", "x3"))
我想要添加新的x4列，但是我在Python中有值，而不是添加到新列中，例如x4_ls = [35.0, 32.0]。是否有最佳方法将新列添加到Spark？(，注意，，我使用Spark2.1)
输出应该类似于：
## +---+---+-----+----+
## | x1| x2| x3|
浏览 0提问于2017-02-15得票数 0
回答已采纳
2回答
我有一个dataframe和如下列的列表：
import spark.implicits._
import org.apache.spark.sql.functions._
val df = spark.createDataFrame(Seq(("Java", "20000"), ("Python", "100000"))).toDF("language","users_count")
val data_columns = List("language","users_c
浏览 9提问于2022-02-08得票数 0
回答已采纳
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
回答已采纳
我试图通过提供XSD模式来解析简单的XML。用这里给出的方法。
XML在这里：
<?xml version="1.0"?>
<beginnersbook>
<to>My Readers</to>
<from>Chaitanya</from>
<subject>A Message to my readers</subject>
<message>Welcome to beginnersbook.com</message>
</beginner
浏览 2提问于2021-04-27得票数 1
我编写了用于从dataframe创建json元素列表的示例spark程序。当我使用main方法执行时，它返回空列表，但是当我执行时没有扩展app的对象时，它返回包含记录的列表。scala对象中扩展App和main方法的区别是什么？
object DfToMap {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.appName("Rnd")
.master("local[*]")
.getOrCreate()
import spar
浏览 1提问于2018-07-17得票数 5
回答已采纳
17回答
如何在Apache Spark DataFrame中连接两列？Spark SQL中有没有我们可以使用的函数？
浏览 7提问于2015-07-16得票数 159
我有两个DataFrame数据帧(A和B)，它们都有一个公共的列/字段(在Spark A中是主键，但在B中不是)。
对于dataframe A中的每一条记录/行，dataframe B中有多条记录。基于该公共列值，我希望针对dataframe A中的每条记录从dataframe B中获取所有记录。
我应该执行什么类型的转换，以便在不做太多混洗的情况下将记录收集在一起？
浏览 0提问于2018-07-25得票数 0
我尝试将spark连接到cassandra，然后从flask对keyspace和table进行查询。 问题是，当我运行web应用程序时，我得到一个错误，告诉我没有创建键空间。cassandra.InvalidRequest: Error from server: code=2200 [Invalid query] message="Keyspace MyKeyspace does not exist" 在spark中，我运行以下命令： val flightRecommendations = finalPredictions.writeStream.foreachBatch {
浏览 30提问于2020-11-29得票数 1
1回答
我试图将oplog.rs从MongoDB加载到spark DataFrame中，它加载了元数据，我通过printSchema函数对其进行了验证，但当我尝试执行像show或count这样的操作时，它给出了这个错误scala.MatchError: ((BsonMinKey,null),0) (of class scala.Tuple2)。我也尝试将其注册为temptable，但仍然给出相同的错误。
val customReadConfig = ReadConfig(Map(
"uri" ->
"mongodb://username:password@h
浏览 0提问于2017-03-04得票数 2
我有多个数据帧需要将它们存储在MapString中，数据帧数据结构。下一步，我们的目标是访问它们以进行连接操作。以下是输入数据帧： names_df:
+-----+----------+----------+
|Id |FirstName | LastName |
+-----+----------+----------+
|1000 | Bob | B |
|1001 | Alice | A |
+-----+----------+----------+
addresses_df
+----+----+----+
|I
浏览 27提问于2019-08-27得票数 0
1回答
我的火花放电版本是2.1.1。我正在尝试连接两个具有两个列( id和优先级)的数据文件(左外部)。我正在创建这样的数据格式：
a = "select 123 as id, 1 as priority"
a_df = spark.sql(a)
b = "select 123 as id, 1 as priority union select 112 as uid, 1 as priority"
b_df = spark.sql(b)
c_df = a_df.join(b_df, (a_df.id==b_df.id), 'left').drop(
浏览 1提问于2019-02-11得票数 2
我正在尝试合并两个不同数据类型的列。在下面的代码片段中，为了简单起见，我从相同的数据帧中选取了列。
from pyspark.sql import SQLContext, Row
from pyspark.sql.types import *
from datetime import datetime
a=sc.parallelize([('ship1',datetime(2015,1,1),2,3.,4.),('ship1',datetime(2015,1,2),4,8.,9.),('ship1',datetime(2015,1,3),5,
浏览 6提问于2015-10-28得票数 4
我试图了解Spark2.0作为一个DataFrame API是如何工作的，DataFrame对数据的结构有一定的了解。
当我把大桌子和小桌子连接起来时，我知道广播小桌子是个好主意。
然而，当加入大表和大表时，有哪些优化技巧？分类有用吗？还是会在内部分拣？什么时候应该重新划分数据？
任何解释都会有帮助
浏览 3提问于2017-06-08得票数 4
*编辑 df_joint = df_raw.join(df_items,on='x',how='left') Apache Spark 2.4.5中出现标题异常 df_raw有2列"x“、"y”的数据，而df_items是模式的空数据框，还有其他一些列 左连接发生在null的值上，它应该从第一个dataframe中获取整个数据，并从第二个dataframe中获取空列。 当"X“是浮点数时，它是完全正常的，但是当我将"X”转换为字符串时，它会抛出隐式笛卡尔乘积的错误 我在spark 2.4.5中收到了这个错误。 为什么会发生这种
浏览 31提问于2020-12-20得票数 0
1回答
我在Spark2.2中打开了几个"csv“文件，但当我做”计数“时，它返回了10000000的记录，而实际上它是6000000的记录，当我用python或Alteryx用Pandas检查它时，它给出了正确的数字。
scala> val df=spark.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema","true").option("encoding",
浏览 4提问于2018-11-27得票数 0
1回答
我需要在spark中实现滞后函数；我可以像下面这样做(使用hive/temp spark表中的一些数据)
假设DF包含以下行：
lagno:value
0, 100
0, 200
2, null
3, null
其中，第一列是要使用的实际滞后数，第二列是实际值。
当我运行这个查询时，它可以工作：
DataFrame df;
DataFrame dfnew=df.select(
org.apache.spark.sql.functions.lag( df.col("value"), 1 ).over(org.apache.spark.sql.express
浏览 3提问于2016-09-16得票数 3
2回答
我有一个数据帧，其中有几个数值列是不固定的(它们在每次执行过程中都会发生变化)。假设我有一个带有数字列名称的Seq对象。我想对这些列中的每一列应用一个聚合函数。我尝试过以下几种方法：
println(numeric_cols)
// -> Seq[String] = List(avgTkts_P1, avgTkts_P2, avgTkts_P3, avgTkts_P4)
var sum_ops = for (c <- numeric_cols) yield org.apache.spark.sql.functions.sum(c).as(c)
var result = df.g
浏览 2提问于2018-09-04得票数 0
我的最终目标是在取一些组的平均值之后，通过组标签得到一个特定的值。要做到这一点，我想获取一个列名的列表，这样我就可以选择具有我感兴趣的组标签的列。但是，生成的DataFrame的行为并不像预期的那样：
import pandas as pd
df = pd.DataFrame({'x': ['a','b','c','a','b','c'], 'y': [1, 2, 3, 4, 5, 6]})
means = df.groupby('x').mean()
浏览 1提问于2019-05-13得票数 0
我想使用Datastax和Spark获得Cassandra表中特定列(列"a")的值和计数的列表，但我在确定执行该请求的正确方法时遇到了麻烦。
我本质上是在尝试做与T-SQL相当的操作
SELECT a, COUNT(a)
FROM mytable
我在Cassandra上使用datastax和spark尝试了以下操作
import com.datastax.spark.connector._
import org.apache.spark.sql.cassandra._
val rdd = sc.cassandraTable("mykeyspace",
浏览 0提问于2017-10-02得票数 1
我想连接由两个列表创建的两个数据帧：
import pandas as pd
import numpy as np
header_1 = ['A', 'B', -1, 3, 5, 7]
data_1 = ['X', 'Y', 1, 2, 3, 4]
d = pd.DataFrame(np.array([data_1]), columns=header_1)
header_2 = ['A', 'B', -2, 4, 5, 6]
data_2 = ['X', 'Z',
浏览 0提问于2016-11-10得票数 1
回答已采纳
很抱歉发表了一篇看似简单的好奇心文章，但我想给出完整的背景.
在Databricks中，我根据特定的模式定义创建“行”数据，然后将该行插入到一个空的dataframe中(也是基于相同的特定模式)。
模式定义如下所示：
myschema_xb = StructType(
[
StructField("_xmlns", StringType(), True),
StructField("_Version", DoubleType(), True),
StructField("MyIds",
ArrayType(
浏览 1提问于2019-12-02得票数 0
回答已采纳
1回答
我有一个自动化SparkSQL作业的用例，我想这样做：
使用火花从菲尼克斯读取一个表(让我们称之为table1)，并收集DataFrame (我们称之为df1)中找到的所有负值
然后，我想从另一个表(table2)中删除记录，其中列的值在df1中(考虑使用连接查询，但我想知道是否可以使用DataFrame，以及是否存在使用HBase和DataFrames的API )。
AFAIK菲尼克斯不直接支持通过星火删除操作(如果我错了，如果有我想知道的方式，请纠正我)，这就是为什么我更倾向于使用HBase Spark
下面是一个可以更直观地解释的模式：
这是一些代码。
浏览 0提问于2019-04-04得票数 2
我有两个spark数据帧df1和df2。有没有一种方法可以在连接这两个数据帧时动态选择输出列？下面的定义在inner join的情况下输出来自df1和df2的所有列。
def joinDF (df1: DataFrame, df2: DataFrame , joinExprs: Column, joinType: String): DataFrame = {
val dfJoinResult = df1.join(df2, joinExprs, joinType)
dfJoinResult
//.select()
}
输入数据：
val df1 = List(("1
浏览 5提问于2018-02-01得票数 3
回答已采纳
1回答
我使用星火在本地模式和一个简单的连接是太长。我获取了两个数据文件:a (8列和230万行)和B(8列和120万行)，并使用A.join(B,condition,'left')连接它们，最后调用了一个操作。它创建一个包含三个阶段的单个作业，每个阶段用于两个数据文件提取，另一个用于加入。令人惊讶的是，提取数据的阶段A大约需要8分钟，而数据B则需要1分钟。几秒钟之内就会加入。我的重要配置设置是：
spark.master本地的*
spark.driver.cores 8
spark.executor.memory 30g
spark.driver.memory 30g
浏览 0提问于2016-12-25得票数 5
回答已采纳
1回答
我不能创建我需要的整个PySpark数据。我现在的字典是这种格式的：
d = {0:
{'Key Features': ['Obese', 'Exercise']},
'Properties': {'Balding': True, 'Tall': False, 'Obese': True, 'Exercise': False}},
1:
{'Key Features': [None]},
'Properties':
浏览 4提问于2021-12-22得票数 0
回答已采纳
我正在尝试创建一个dataframe，其中第一列是令牌列表，并且可以在其中添加附加的信息列。但是，pandas不允许将标记列表添加为一列。因此，代码如下所示
array1 = ['two', 'sample', 'statistical', 'inferences', 'includes']
array2 = ['references', 'please', 'see', 'next', 'page', 'the'
浏览 0提问于2016-05-12得票数 0
我使用星火与纱线簇管理器。
我从Cassandra表中创建了一个数据集，该表包含大约700行数据，其中5列包含5列，其中一列具有JSON格式的数据。数据量仅以MB为单位。
我运行spark-shell时：
spark.executor.memory=4g
spark.driver.memory=2g
我得到了这个错误：
org.apache.spark.shuffle.FetchFailedException:未能连接到bosirahtaicas02.bscuat.local 02. org.apache.spark.storage.ShuffleBlockFetcherIt
浏览 7提问于2017-05-31得票数 3
我使用的是kafka connect cassandra源连接器1.0版本。我在cassandra表中有一个十进制数据类型列(价格)，并将它作为json从源连接器写入到kafka主题中，它正在以某种字符串格式写入小数值，例如"price":"AA=="。现在它在我的spark流中给出错误，同时转换为float作为“数字格式异常”....？？请建议在写kafka主题中的值时可能出错的地方。提前谢谢。
浏览 2提问于2018-11-20得票数 0
1回答
PySpark数据显示错误的值
、、、、
我刚刚从Pandas转到了PySpark dataframe，发现在PySpark dataframe中打印出相同的列会给出错误的值。下面是一个例子:使用Pandas：
df_pandas=pd.read_csv("crime.csv", low_memory=False)
print(df_pandas["CRIMEID"].head(5))
Output:
1321797
1344185
1181882
1182632
1195867
而使用PySpark数据帧：
df_spark = sqlContext.read.format('csv'
浏览 1提问于2018-02-27得票数 0
回答已采纳
按照官方的，我们不能向DataFrame添加特定的单元分区
直至Spark1.5，以下内容用于工作，而dataframe将有实体列和数据，如下所示：
DataFrame df = hiveContext.read().format("orc").load("path/to/table/entity=xyz")
然而，这将不适用于星火1.6。
如果给出如下所示的基路径，它不包含我希望在DataFrame中包含的实体列，如下所示-
DataFrame df = hiveContext.read().format("orc").load("pa
浏览 3提问于2016-01-07得票数 7
回答已采纳
1回答
当我从s3桶将数据加载到pyspark中时，进行一些操作(连接、联合)，然后尝试覆盖前面读取的相同路径(' data /csv/')。我得到了一个错误：
py4j.protocol.Py4JJavaError: An error occurred while calling o4635.save.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.sca
浏览 0提问于2019-06-12得票数 0
我有一个连接到mongodb和几个数据库和collecions在里面，我只想有一个连接，使查询到几个集合在pyspark。我认为每个查询都有一个连接延迟了性能。
这就是我所拥有的
database_1 = "data_1"
database_2 = "data_2"
collection_1 = "client_1"
collection_2 = "client_2"
myquery_1 = [query_1]
myquery_2 = [query_2]
dataframe_1 = spark.read.format("
浏览 1提问于2022-11-16得票数 0
1回答
crossJoin可以按以下方式完成：
df1 = pd.DataFrame({'subgroup':['A','B','C','D']})
df2 = pd.DataFrame({'dates':pd.date_range(date_today, date_today + timedelta(3), freq='D')})
sdf1 = spark.createDataFrame(df1)
sdf2 = spark.createDataFrame(df2)
sdf1.crossJ
浏览 0提问于2019-04-19得票数 2
回答已采纳
1回答
我有一个DataFrame (转换为RDD)，并希望重新分区，以便每个键(第一列)都有自己的分区。这是我所做的：
# Repartition to # key partitions and map each row to a partition given their key rank
my_rdd = df.rdd.partitionBy(len(keys), lambda row: int(row[0]))
但是，当我试图将它映射回DataFrame或保存它时，我得到了这个错误：
Caused by: org.apache.spark.api.python.PythonException:
浏览 0提问于2015-11-21得票数 5
在“火花:最终指南”中，它说：
如果需要引用特定DataFrame的列，则可以在特定的DataFrame上使用col方法。
例如(在Python/Pyspark中)：
df.col("count")
但是，当我在包含列count的dataframe上运行后的代码时，就会得到错误'DataFrame' object has no attribute 'col'。如果我尝试column，我会得到一个类似的错误。
这本书是错的，还是我该怎么做呢？
我上的是星火2.3.1。dataframe是用以下内容创建的：
df = spark.read.f
浏览 2提问于2018-08-12得票数 9
我正在试着把一个火花DataFrame转换成熊猫。但是，它给出了以下错误：
OutOfBoundsDatetime: Out of bounds nanosecond timestamp:
有没有解决这个问题的办法？
如果我删除了所有的时间戳列，那么它就可以工作，但是我想把整个表都放到Pandas中。
当我把Spark DataFrame带到Pandas时，我从来没有遇到过错误。
这是一个相当大的表，其中包含多个时间戳列。有些是YYYY-MM-DD列，有些是YYYY-MM-DD 00:00:00类型的列。
有几列包含不存在的年份变量，数量不详。
下面是一个例子。
data = {
&#
浏览 140提问于2021-02-23得票数 2
回答已采纳
我有两个数据帧，第一个数据帧包含实际数据(从CSV文件读取)，第二个数据帧包含一个具有多个关键字的列。即下面的Dataframe-1外观相似(我们必须搜索的地方)：
数据帧-2：
我想要的输出：
我在这里使用spark scala。我想要一个与dataframe-1中的dataframe-2完全匹配的单词。我使用了like、rlike、contains等函数，但它没有给出我想要的输出。有人知道如何在spark scala SQL中或使用dataframe的spark scala函数来开发此逻辑吗？请帮我弄一下这个。
浏览 1提问于2021-02-12得票数 0
我使用的是Spark 1.3.1，其中连接两个数据帧会重复连接的列。我在外部连接两个数据帧，希望将结果数据帧发送到na().fill()方法，以便根据列的数据类型将空值转换为已知值。我构建了一个"table.column“-> "value”的映射，并将其传递给fill方法。但我得到的是异常而不是成功:(。我有什么选择？我看到有一个dataFrame.withColumnRenamed方法，但我只能重命名一列。我有涉及多个列的连接。我是否只需要确保存在一组惟一的列名，而不管我应用na().fill()方法的dataFrame中的表别名是什么？
给定：
scala>
浏览 4提问于2016-02-28得票数 3
在Spark中，将Timestamp列转换为毫秒时间戳Long列的最短和最有效的方法是什么？
下面是一个从时间戳到毫秒的转换示例
scala> val ts = spark.sql("SELECT now() as ts")
ts: org.apache.spark.sql.DataFrame = [ts: timestamp]
scala> ts.show(false)
+-----------------------+
|ts
浏览 1提问于2019-06-18得票数 2
1回答
我有C*列家族来存储类似事件的数据。以这种方式在CQL3中创建的列族：
CREATE TABLE event (
hour text,
stamp timeuuid,
values map<text, text>,
PRIMARY KEY (hour, stamp)
) WITH CLUSTERING ORDER BY (stamp DESC)
分区器是Murmur3分区器。然后，我尝试通过库构建对该数据的Spark。结果，我遇到了两个问题：
在我的例子中，集群键(“hour”字段)有1000多条记录，但是响应每个键只包含头1000条记录。我可以在查询中增加页面
浏览 6提问于2014-03-06得票数 1
回答已采纳
2回答
我在数据框中执行聚合以计算百分比。我需要将每列的总和存储在一个单独的变量中，我可以在除法中使用它来计算百分比
val sumOfCol1 = df.agg(round(sum("col1"),2))
这段代码给出了和，但它将被存储为dataframe对象，并且不能用于除法。类型为：
sumOfCol1: org.apache.spark.sql.DataFrame = [round(sum(col1), 2): double]
如何将其存储为常量或双精度值，以便在聚合的后期阶段使用它？
浏览 37提问于2019-05-25得票数 -1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
