当我使用带有spark/scala的窗口时，我不能改变分区吗？
当使用带有Spark/Scala的窗口时，可以改变分区。窗口函数是一种在数据流中执行聚合操作的方式，它将数据划分为不同的窗口，并在每个窗口上执行聚合操作。在Spark中，可以使用窗口函数来对数据进行分组、排序和聚合。
在Spark中，可以使用partitionBy方法来改变分区。partitionBy方法接受一个或多个列名作为参数，并根据这些列名将数据进行分区。分区可以根据数据的某些特征进行划分，以便在执行窗口函数时可以更高效地处理数据。
以下是使用Spark/Scala进行窗口操作的示例代码：
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
val df = spark.read.format("csv").load("data.csv") // 从CSV文件加载数据
val windowSpec = Window.partitionBy("column1", "column2").orderBy("column3") // 定义窗口规范
val result = df.withColumn("sum_column4", sum("column4").over(windowSpec)) // 在窗口上执行聚合操作
result.show()在上述示例中，我们使用partitionBy方法将数据按照"column1"和"column2"进行分区，并按照"column3"进行排序。然后，我们使用sum函数在窗口上计算"column4"的总和，并将结果保存在新的列"sum_column4"中。最后，我们使用show方法显示结果。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark服务
- 腾讯云Scala开发工具包
- 腾讯云数据仓库
- 腾讯云大数据计算服务
- 腾讯云数据分析平台
- 腾讯云数据湖分析
- 腾讯云数据集成服务
- 腾讯云数据传输服务
- 腾讯云数据备份与恢复
- 腾讯云数据加密服务
- 腾讯云数据安全服务
- 腾讯云数据治理服务
- 腾讯云数据质量管理
- 腾讯云数据集市
- 腾讯云数据可视化
- 腾讯云数据分析与挖掘
- 腾讯云数据标注服务
- 腾讯云数据集成开发环境
- 腾讯云数据开发工具
- 腾讯云数据迁移工具
- 腾讯云数据同步服务
- 腾讯云数据流计算服务
- 腾讯云数据仓库服务
- 腾讯云数据查询服务
- 腾讯云数据集成与分析
- 腾讯云数据分析与挖掘平台
- 腾讯云数据开发工具包
- 腾讯云数据集成开发工具包
- 腾讯云数据迁移工具包
- 腾讯云数据同步工具包
- 腾讯云数据流计算工具包
- 腾讯云数据仓库工具包
- 腾讯云数据查询工具包
- 腾讯云数据集成与分析工具包
相关·内容
当我在一个80节点的集群上运行Azure HDInsight 3.6时，我在Spark 2.3中遇到了这个异常： java.lang.UnsupportedOperationException: Can not build a HashedRelation that is larger than 8G
at org.apache.spark.sql.execution.joins.LongToUnsafeRowMap.grow(HashedRelation.scala:623)
at org.apache.spark.sql.execution.joins.LongToUnsafeRowMa
浏览 44提问于2020-07-02得票数 0
回答已采纳
1回答
我正面临一个问题，我的火花工作被困在当地，而运行在IntelliJ的想法。我的工作一直持续到一个阶段，比如完成200个任务中的199个，或者完成3个任务中的一个，然后被困在那里。
我试图在IDE中使用评估表达式查看正在发生的事情，并注意到了一个奇怪的问题。如果我使用的是myDf.rdd.map(r => r).cache()
java.io.IOException: Class not found
at org.apache.xbean.asm5.ClassReader.a(Unknown Source)
at org.apache.xbean.asm5.ClassReader.<
浏览 1提问于2018-03-15得票数 1
1回答
当我执行以下命令时：
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <console>:22
scala> rdd.partitions.size
res9: Int = 10
scala> rdd.partitioner.isDefined
浏览 1提问于2015-12-28得票数 20
回答已采纳
我正在尝试在JMSTemplate方法中使用Spring rdd.foreach类，但是我得到的是任务而不是可序列化的错误。当我尝试使用静态变量时，它在本地运行，但在集群中，我得到的是空指针异常。
示例代码：
inputRDD.foreach(record -> {
messageServices.send(record);
}
错误日志：
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.
浏览 3提问于2015-12-29得票数 0
回答已采纳
我有一个RDD，它是通过读取一个大小约为117 of的本地文本文件形成的。
scala> rdd
res87: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at textFile at <console>:24
我缓存RDD:-
scala> rdd.persist()
res84: rdd.type = MapPartitionsRDD[3] at textFile at <console>:24
在此之后，我将RDD上的‘采取(1)’操作称为强制求值。完成后，我将检查Spark的存储页面。
浏览 4提问于2015-07-20得票数 3
我正在使用UR，我想知道，为什么在培训期间大多数时间只使用4个内核，而8个可用
大部分时间的培训都停留在工作17 "take at EsSpark.scala:60“Stage45 "flatMap at AtB.scala:234”
有人能解释一下它在这个阶段到底是做什么的吗?它可以使用所有8个内核(不仅仅是4个)吗？
我想，它没有足够的RAM，因为它使用了所有可用的64G RAM，但当我在spark集群(2个实例)上运行它时，情况没有改变，它总共也使用了4个内核。
浏览 0提问于2016-11-12得票数 0
各位，
当我试图在每行一个键标识符上加入2个大数据(100 to +每个)时，我遇到了这个问题。
我在EMR上使用Spark1.6，下面是我正在做的事情：
val df1 = sqlContext.read.json("hdfs:///df1/")
val df2 = sqlContext.read.json("hdfs:///df2/")
// clean up and filter steps later
df1.registerTempTable("df1")
df2.registerTempTable("df2")
浏览 1提问于2016-11-25得票数 7
在设置FetchFailedException = 2700时，当我加入到表中时，我会遇到一个spark.sql.shuffle.partitions
但是，当设置spark.sql.shuffle.partitions = 500时，可以成功运行。
如我所知，增加shuffle.partitions将减少每次任务中的数据洗牌时。
我错过了什么吗？
异常
FetchFailed(BlockManagerId(699, nfjd-hadoop02-node120.jpushoa.com, 7337, None), shuffleId=4, mapId=59, reduceId=1140, mes
浏览 6提问于2020-06-14得票数 0
6我尝试实现了所解释的东西。当我在自定义分区中保持分区数等于1时，它是有效的，但当我更改它时，保留任何其他值，它会给出超出边界的数组异常
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6, deenbandhu): java.lang.ArrayIndexOutO
浏览 2提问于2016-06-10得票数 2
我正在尝试修复在我的spark设置中看到的一个外存问题，在这一点上，我无法得出一个具体的分析，为什么我会看到这个。在将数据帧写到parquet或kafka时，我总是会遇到这个问题。我的数据帧有5000行。它的模式是
root
|-- A: string (nullable = true)
|-- B: string (nullable = true)
|-- C: string (nullable = true)
|-- D: array (nullable = true)
| |-- element: string (contains
浏览 0提问于2017-09-14得票数 0
1回答
我需要将数据集数据划分为6列: region/year/month/day/id/quadkey，其中在顶层我只有二进制区域状态，而在最底层实际上是它进入许多分区的位置。假设我们有两个区域/通常是一年/通常是一个月/3-4天/100-150个I/ 50-200个四键，当我执行这个操作时，我得到了非常不平衡的随机操作，有时执行器会因为超出内存限制而失败。此外，我从History UI中注意到，hat阶段的一些任务非常大(约15 1Gb)，而其他任务则小得多(约1 1Gb)。
我试着去玩
sqlContext.setConf("spark.sql.shuffle.partitions
浏览 0提问于2019-04-11得票数 1
最近，我们开始面对spark 2.4.4与Hive1.2.1版本的问题。
当我们试图从按字符串类型进行分区的表中读取数据时
spark.sql("select count(*) from table where #conditions");
Caused by: java.lang.RuntimeException: Caught Hive MetaException attempting to get partition metadata by filter from Hive. You can set the Spark configuration setting spa
浏览 3提问于2021-07-13得票数 0
回答已采纳
我正在尝试使用RDD ()方法来更改RDD的分区大小。对RDD的方法调用成功，但是当我使用RDD的partition.size属性显式检查分区大小时，我得到的分区数目与它原来的相同：-
scala> rdd.partitions.size
res56: Int = 50
scala> rdd.repartition(10)
res57: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[19] at repartition at <console>:27
在这个阶段，我执行一些像rdd.take(1)这样的操作，只
浏览 2提问于2015-07-20得票数 5
回答已采纳
3回答
我在Scala中有一个程序，它读取一个CSV文件，向Dataframe添加一个新列，并将结果保存为一个parquet文件。它在小文件(<5 Go)上工作得很好，但是当我尝试使用更大的文件(~80 Go)时，它总是会失败，因为它应该用这个堆栈跟踪来编写拼花文件：
16/10/20 10:03:37 WARN scheduler.TaskSetManager: Lost task 14.0 in stage 4.0 (TID 886, 10.0.0.10): java.io.EOFException: reached end of stream after reading 136445 by
浏览 0提问于2016-10-20得票数 1
我现在正在笔记本电脑上使用Spark和Scala。
当我将RDD写入文件时，输出将被写入两个文件"part-00000“和"part-00001”。如何强制Spark / Scala向写入一个文件
我的代码是：
myRDD.map(x => x._1 + "," + x._2).saveAsTextFile("/path/to/output")
我在哪里写键，值对。
浏览 4提问于2016-02-17得票数 5
回答已采纳
我正在尝试从多个路径读取一些avro文件到DataFrame。假设我的路径是这个路径下的"s3a://bucket_name/path/to/file/year=18/month=11/day=01"，我还有两个分区，假设是country=XX/region=XX
我想一次读取多个日期，而不需要明确命名国家和地区分区。此外，我希望国家和地区是这个DataFrame中的列。
sqlContext.read.format("com.databricks.spark.avro").load("s3a://bucket_name/path/to/file/yea
浏览 2提问于2018-12-03得票数 2
请看下面的代码。当我传递分区数的值时，下面的代码会出错。
def loadDataFromPostgress(sqlContext: SQLContext, tableName: String,
columnName: String, dbURL: String, userName: String, pwd: String,
partitions: String): DataFrame = {
println("the no of partitions are : "+partitions)
v
浏览 0提问于2017-04-15得票数 0
回答已采纳
当我为表同步运行spark应用程序时，错误消息如下所示：
19/10/16 01:37:40 ERROR Executor: Exception in task 0.0 in stage 3.0 (TID 51)
com.mysql.cj.jdbc.exceptions.CommunicationsException: Communications link failure
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packet
浏览 3提问于2019-10-16得票数 0
回答已采纳
2回答
我希望将行号(row_num)创建为MySql中现有表的列，通过spark并行读取数据库(即分区列，因为表中的所有列都是字符串)。
当我试图执行这个查询时：
val query = SELECT @row_number:=@row_number+1 as rowid,d.* FROM destination d, (SELECT @row_number:=0) as init
我有一个例外情况如下：
17/10/16 10:50:00 INFO SparkSqlParser: Parsing command: SELECT @row_number:=@row_number+1 as rowid
浏览 8提问于2017-10-16得票数 0
回答已采纳
1回答
我正在尝试使用以下配置在纱线集群中执行星火作业。
/usr/bin/spark-submit
--class com.example.DriverClass
--master yarn-cluster
app.jar
hdfs:///user/spark/file1.parquet
hdfs:///user/spark/file2.parquet
hdfs:///user/spark/output
20151217052915
--num-executors 20
--executor-memory 12288M
--executor-cores 5
--driver-m
浏览 2提问于2015-12-19得票数 2
我正在阅读，关于分区，他说
默认情况下，为每个HDFS分区创建一个分区，默认为64 is。
我对HDFS并不十分熟悉，但是我遇到了一些复制这条语句的问题。我有一个名为Reviews.csv的文件，它是大约330 of的亚马逊食品评论文本文件。给定默认的64‘d块，我希望使用ceiling(330 / 64) = 6分区。但是，当我将文件加载到我的Shell中时，我得到了9个分区：
scala> val tokenized_logs = sc.textFile("Reviews.csv")
tokenized_logs: org.apache.spark.rdd.R
浏览 1提问于2018-12-01得票数 1
回答已采纳
1回答
我在Apache上使用，它有4个节点，总共210 GB。
我的星火作业是在一个小的事务数据集和一个大型事件数据集之间进行关联。我想根据时间和ID (事件时间和事务时间、ID和ID)将每个事务与最近的事件匹配。
我得到以下错误：
FetchFailed(null, shuffleId=1, mapId=-1, reduceId=20,
message=org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 1
at org.apache.spark.Ma
浏览 2提问于2016-03-10得票数 2
我有一个数据，并写入到S3桶目标位置。在代码中，合并用于加载数据和获取SparkOutOfMemoryError.The当前代码，这些代码使用了多个项目，并且看到了许多推荐的重新分区解决方案，它对我起了作用。即使它没有记录，合并也不起作用。有没有其他的方法来解决这个问题而不改变为重新划分？
代码：
empsql = 'Select * From Employee'
df = spark.sql(empsql) ##Spark is configured
df.coalesce(2).write.mode('overwrite').format("parq
浏览 8提问于2020-06-11得票数 2
我使用SparkSQL处理数据，并且我想将我的数据写入一个子文件。
...
step12.show()
step12.repartition(10).coalesce(1).write.json('wasb://liu@cliubo.blob.core.windows.net/test_data_4')
step12是我的数据帧，但是我收到一个错误，告诉我java.lang.OutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0它没有意义，因为我可以显示这个数据帧。我使用Microsoft Azure中的
浏览 0提问于2017-04-17得票数 0
我正在尝试使用Spark在S3事件上创建一个简单的sql查询。我正在加载大约30 as的JSON文件，如下所示：
val d2 = spark.read.json("s3n://myData/2017/02/01/1234");
d2.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK);
d2.registerTempTable("d2");
然后，我尝试将查询结果写入文件：
val users_count = sql("select count(distinct data.use
浏览 1提问于2017-02-15得票数 26
回答已采纳
当我通过rdd.repartition(1).saveAsTextFile(file_path)保存一对rdd时，会遇到一个错误。
Py4JJavaError: An error occurred while calling o142.saveAsTextFile.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.internal.io.SparkHadoopWriter$.write(SparkHadoopWriter.scala:100)
at org.apache.spark.rdd.
浏览 1提问于2020-07-01得票数 0
回答已采纳
我们使用Apache Beam，它是在火星车上执行的。我们的案件如下。这两个用例都会导致OutofMemory错误。
1)连接-2个使用Apache的大表--一个大小为120 is的表，另一个为60 is的表。这会在OutofMemory ()内部调用groupByKeyOnly()时导致GroupCombineFunctions.java错误。
2) GroupByKey --我们根据如下键对数据集进行分组。PCollection>> costBasisRecords = masterDataResult.apply(GroupByKey.create())；
此GroupbyK
浏览 1提问于2017-12-16得票数 0
回答已采纳
1回答
我很喜欢使用Spark，但就在此之前我遇到了一个问题。由于RDD的每个分区的大小限制，Spark会产生以下错误消息；(INT_MAX)。
16/03/03 15:41:20 INFO TaskSetManager: Lost task 142.2 in stage 4.0 (TID 929) on executor 10.1.100.116: java.lang.RuntimeException (java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
at sun.nio.ch.FileChannelImp
浏览 5提问于2016-03-03得票数 1
回答已采纳
我读到具有相同分区的RDDs将被放在同一位置。这对我来说很重要，因为我想连接几个未分区的大型Hive表。我的理论是，如果我可以对它们进行分区(通过一个名为date_day的字段)并将它们放在一起，那么我就可以避免混洗。
下面是我尝试为每个表做的事情：
def date_day_partitioner(key):
return (key.date_day - datetime.date(2017,05,01)).days
df = sqlContext.sql("select * from hive.table")
rdd = df.rdd
rdd2 = rdd.parti
浏览 2提问于2017-11-05得票数 5
我使用以下命令在本地模式下使用Spark2.0调用Pyspark：
pyspark --executor-memory 4g --driver-memory 4g
输入数据正在从tsv文件中读取，并且有580 Kx28列。我正在对dataframe做一些操作，然后我尝试将它导出到tsv文件中，我得到了这个错误。
df.coalesce(1).write.save("sample.tsv",format = "csv",header = 'true', delimiter = '\t')
任何关于如何消除这个错误的提示。我可以很容易
浏览 7提问于2016-08-15得票数 24
回答已采纳
我使用以下代码在星火GraphX中创建了一个图形。(见)
import scala.math.random
import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.util.Random
import org.apache.spark.HashPartitioner
object SparkER {
val nPartitions: Integer = 4
val n: Long = 100
val p: Double = 0.
浏览 0提问于2015-12-22得票数 3
回答已采纳
我的工作是火花1.6，它失败了我的工作与以下错误
/data/05/dfs/dn/yarn/nm/usercache/willir31/appcache/application_1413512480649_0108/spark-local-20141028214722-43f1/26/shuffle__312_0.index (无此类文件或目录)java.io.FileOutputStream.open(原生方法) java.io.FileOutputStream.(FileOutputStream.java:221) org.apache.spark.storage.DiskBlockOb
浏览 3提问于2016-07-14得票数 4
我正在使用Google Cloud Dataproc做spark工作，我的编辑器是Zepplin。我正在尝试将json数据写入gcp存储桶。当我尝试10MB文件时，它成功了。但由于文件为10 But而失败。我的dataproc有一个主服务器，4CPU，26 My内存，500 My磁盘。5个具有相同配置的工作进程。我猜它应该能够处理10 it的数据。
我的命令是toDatabase.repartition(10).write.json("gs://mypath")
错误是
org.apache.spark.SparkException: Job aborted.
at org
浏览 4提问于2019-04-08得票数 11
回答已采纳
1回答
当我们在Hortonworks集群上通过木星笔记本处理以下步骤时，我们遇到了一个奇怪的情况：
从火花放电数据库中的ORC表中读取数据
用pivot_column (pivoted_df)旋转这个表
在pivoted_df：calced_df = pivoted_df.select(dependency_list).run_calculations()的特定选择上添加一些计算过的列
在大表pivoted_df (列> 1.600)和“小”表calced_df (仅~270个列)上进行内部连接，以合并所有列
保存到Parquet桌
(在步骤3中，选择是必要的，否则
浏览 2提问于2018-08-06得票数 1
我有一个火花工作，不断上传Parquet文件到S3 (与分区)。
这些文件都有相同的拼花模式。
其中一种字段类型最近已被更改(从String改为long)，因此某些分区的拼花模式是混合的。
来自这两种类型的混合数据的地方现在失败了，无法读取某些内容。
虽然我似乎可以执行：sqlContext.read.load(path)
尝试在DataFrame上应用任何提取操作(例如，collect)时，该操作在ParquetDecodingException中失败
我打算迁移数据并将其重新格式化为，但无法将混合内容读取到DataFrame中。
如何使用Apache将混合分区加载到DataFrames或任
浏览 0提问于2018-06-21得票数 0
回答已采纳
当我运行一个带有火花的模型训练管道时，会出现上述错误。
`val inputData = spark.read
.option("header", true)
.option("mode","DROPMALFORMED")
.csv(input)
.repartition(500)
.toDF("b", "c")
.withColumn("b", lower(col("b")))
.withColumn("c", lower(col(
浏览 2提问于2018-06-28得票数 2
我尝试使用汇流平台，并以为例，向REST端点发出高级别的Kafka请求。
我使用以下Kafka参数：
val kafkaParams = Map(
"bootstrap.servers" -> "localhost:9092",
"schema.registry.url" -> "http://localhost:8081",
"group.id" -> "EventConsumer",
"auto.offset.reset" -> "
浏览 3提问于2016-09-05得票数 0
我想出了四种方法来解决这个问题，但没有一种方法对大型RDD有效，我希望有人能帮我解决。
我有一个格式为((x，y)，( sim，sim'))的Spark RDD，其中x和y是两个索引，sim和sim‘是x和y的两个不同的相似性度量。我有兴趣寻找具有最大sim’值的元组。我想出了几种方法来做到这一点，但每种方法都有自己的问题，最终这些方法都不能应用于大型RDD，例如10亿个元组的RDD。
假设res_dict是((x，y)，(sim，sim'))元组的RDD，当在pyspark中调用res_dict.collect()时，它返回[((0, 4), (0.8448286521635
浏览 0提问于2016-10-14得票数 2
我尝试在带有hdfs的YARN集群上运行一个用Scala编写的Spark/Graphx程序。群集有16个节点，每个节点具有16 HD和2TB硬盘。我想要的是使用edgeListFile函数加载一个3.29 is的无向图(称为orkutUndirected.txt)，该函数由GraphX库提供：
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.streaming._
import org.apache.spark.streaming.Stream
浏览 1提问于2016-02-27得票数 1
2回答
我正在尝试对消费者群体进行实验。
这是我的代码片段
public final class App {
private static final int INTERVAL = 5000;
public static void main(String[] args) throws Exception {
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "xxx:9092");
kafka
浏览 0提问于2018-05-27得票数 2
回答已采纳
我想创建一个带有分区的hive表。
该表的模式为：
val schema = StructType(StructField(name,StringType,true),StructField(age,IntegerType,true))
我可以使用Spark-SQL来做这件事，使用：
val query = "CREATE TABLE some_new_table (name string, age integer) USING org.apache.spark.sql.parquet OPTIONS (path '<some_path>') PARTITI
浏览 7提问于2017-08-13得票数 1
我在斯派克工作，一路走来接斯卡拉。我有一个关于RDD api以及如何实现各种基本RDD的问题。具体来说，我在火花壳中运行了以下代码：
scala> val gspeech_path="/home/myuser/gettysburg.txt"
gspeech_path: String = /home/myuser/gettysburg.txt
scala> val lines=sc.textFile(gspeech_path)
lines: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[7]
at text
浏览 4提问于2016-01-29得票数 1
回答已采纳
1回答
我有以下卡桑德拉·DM：
CREATE TABLE table (
id uuid,
timestamp timestamp STATIC,
value1 text STATIC,
value2 int,
value3 text,
data map <text,text>,
PRIMARY KEY ( id, value2 )
);
到目前一切尚好。现在，我必须先插入类型的格式：
INSERT INTO table (id, timestamp, value1)
VALUES (<uuid>,<times
浏览 3提问于2017-01-20得票数 1
回答已采纳
1回答
我收到一个错误，从我在这里找到的相关问题中，答案很好地可能是增加分区数。我被困的地方实际上是实现解决方案。
三个问题：
(A)是否需要回过头来实例化星火上下文并在其中定义分区号？示例代码将不胜感激。
(B)更好的是，我是否可以“就位”增加分区？有些代码行需要很长时间才能执行；如果可能的话，我希望避免重新开始。
(C)也许我应该先问这个问题；增加分区数目是否解决以下错误的适当办法？
users = r.select('user_id').distinct().collect()
--------------------------------------------------
浏览 2提问于2019-11-21得票数 1
回答已采纳
我正在使用RandomForestCLassifier构建一个模型。这是我的密码
conf = SparkConf()
conf.setAppName('spark-nltk')
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
m=sc.textFile("Question_Type_Classification_testing_purpose/data/TREC_10.txt").map(lambda s: s.split(" ",1))
df= m.toDF()
创建的data
浏览 6提问于2017-04-01得票数 3
回答已采纳
我有一个Hadoop集群，它使用Apache来查询保存在Hadoop上的parquet文件。例如，当我使用下面的PySpark代码在拼花文件中找到一个词时：
df = spark.read.parquet("hdfs://test/parquets/*")
df.filter(df['word'] == "jhon").show()
在运行这段代码之后，我将转到stages应用程序UI，stages选项卡。与此形成对比的是，由于该查询的性质，它必须在本地运行，至少在Any.本地级上运行。当我在运行时检查集群的网络IO时，我发现这个查询使用网络(当
浏览 8提问于2021-03-13得票数 1
回答已采纳
对于spark，我们应该如何从HDFS中的一个文件夹中获取数据，进行一些修改，并通过覆盖保存模式将更新后的数据保存到HDFS 中的同一个文件夹中，而不需要FileNotFoundException。
import org.apache.spark.sql.{SparkSession,SaveMode}
import org.apache.spark.SparkConf
val sparkConf: SparkConf = new SparkConf()
val sparkSession = SparkSession.builder.config(sparkConf).getOrCreate(
浏览 4提问于2017-03-21得票数 18
所以，我有如下数据，
[ (1, data1), (1, data2), (2, data3), (1, data4), (2, data5) ]
我想将其转换为以下内容，以供进一步处理。
[ (1, [data1, data2, data4]), (2, [data3, data5]) ]
我使用了groupByKey和reduceByKey，但是由于数据量很大，它失败了。数据不是很高，而是很宽。换句话说，键在1到10000之间，但是值列表从100 K到900 k不等。
我正在努力解决这个问题，并计划应用mapPartitions或(Hash)partitioner。
所以，如果其中一个可行
浏览 5提问于2016-01-26得票数 6
我有一个数据帧，我希望根据现有的日期列按年、月和日进行分区。最后，我想把分区的csv写到某个目录中。 但是在写入目录时，当我调用partionBy()函数时，它会抛出下一个异常。当使用coalesce()时，我可以像预期的那样写入目录。 以下是pyspark版本2.3的代码片段 import pyspark.sql.functions as func
df1 = flights.select("airlines","date")\
.withColumn('year' , func.year(flight
浏览 32提问于2019-03-02得票数 1
回答已采纳
我正在查看RDD的分区
>>> rdd = sc.parallelize(range(10), 5)
>>> rdd.collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> rdd.glom().collect()
[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]
>>> rdd.mapPartitions(lambda x: [x]).collect()
[range(0, 2), range(2, 4), range(4, 6), range(6
浏览 2提问于2018-06-28得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
