apache-spark 相关问题

通过 PySpark 将 2 GBPS 数据推送到 Kafka 主题的最佳方式

我正在尝试对 Kafka Producer 和 Kafka Consumer 进行负载测试。因此，我生成了一个文件的 2 GBPS 数据，每个文件的有效负载为 4KB。我已经使用 PySpark 加载这些数据...

pyspark——对 Array(Integer()) 类型的列中的值求和的最佳方法

可以说这是我的数据框...... 名称 |分数 丹| [10,5,2,12] 安| [ 12,3,5] 乔恩 | [ ] 期望的输出是这样的 名称 |分数 |全部的 丹| [10,5,2,12] | 29 安...

如何优化Azure Synapse Spark笔记本中大数据集的累积和字段计算？

我有一个包含 3M+ 条记录和几列的数据集。这是我的数据集的示例： 物品 项目库 日期 数量_1 数量_2 1 20 202410 600 7493 1 20 202411 17000 16431 每个item-item_base...

Hadoop：线程“main”中出现异常 java.lang.UnsupportedOperationException：不支持“posix：permissions”作为初始属性

C:\Users\sudha>hadoop jar "C:\hadoop\share\hadoop\mapreduce\hadoop-mapreduce-examples-3.4.0.jar" wordcount /newdir/HadoopSmall.txt /newdir/smallword 我正在使用这个命令来处理单词 co...

将新列附加到现有镶木地板文件

有什么方法可以将新列附加到现有的镶木地板文件中吗？ 我目前正在参加 Kaggle 比赛，我已将所有数据转换为镶木地板文件。 情况就是这样，我读了

如何在Python程序中管理数百个相关指标的计算？

为了计算医院科室的绩效工资，我们会涉及到很多指标，比如：门诊量、床位数、收入、支出等；这些指标是ab...

无法从apache Spark连接到cassandra：com.datastax.oss.driver.api.core.connection.ClosedConnectionException：与远程对等点失去连接

[从 docker windows 运行的 cassandra] 我正在从 wsl2 运行 Spark Spark-shell --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1 [上面命令后的 Spark-shell] ...

Spark Scala：java.lang.NoSuchFieldError：JAVA_9 位于 org.apache.spark.storage.StorageUtils$。<init>（StorageUtils.scala：207）

我有一个spark scala应用程序，它使用列出的以下版本（仅粘贴pom.xml的一部分）的依赖项和属性。 依赖项： org.apache.spark 我有一个 Spark scala 应用程序，它使用以下版本（仅粘贴 pom.xml 的一部分）列出的依赖项和属性。 依赖关系： <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.12</artifactId> <version>3.3.2</version> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>3.3.2</version> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-reflect</artifactId> <version>2.12.10</version> <scope>compile</scope> </dependency> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-compiler</artifactId> <version>2.12.10</version> <scope>compile</scope> </dependency> 特性： <scala.major.version>2.12</scala.major.version> <scala.test.version>3.2.18</scala.test.version> <scala.version>2.12.18</scala.version> <maven.assembly.plugin.verion>3.3.0</maven.assembly.plugin.verion> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <maven.deploy.plugin.version>2.8.2</maven.deploy.plugin.version> <maven.properties.plugin.verion>1.0.0</maven.properties.plugin.verion> <maven.release.plugin.version>3.0.0</maven.release.plugin.version> <maven.scala.plugin.version>4.5.6</maven.scala.plugin.version> <maven.shade.plugin.version>3.5.2</maven.shade.plugin.version> <maven.site.plugin.version>3.12.1</maven.site.plugin.version> <maven.source.plugin.version>3.2.1</maven.source.plugin.version> 我的本地设置有 java=1.8.0_202 和 mvn=3.9.6 我能够成功编译 src/main 和 src/test --> “mvn test-compile”。 当我运行“mvn install”时，单元测试开始失败，并出现以下错误。 *** RUN ABORTED *** An exception or error caused a run to abort: JAVA_9 java.lang.NoSuchFieldError: JAVA_9 at org.apache.spark.storage.StorageUtils$.<init>(StorageUtils.scala:207) at org.apache.spark.storage.StorageUtils$.<clinit>(StorageUtils.scala) at org.apache.spark.storage.BlockManagerMasterEndpoint.<init>(BlockManagerMasterEndpoint.scala:114) at org.apache.spark.SparkEnv$.$anonfun$create$9(SparkEnv.scala:353) at org.apache.spark.SparkEnv$.registerOrLookupEndpoint$1(SparkEnv.scala:290) at org.apache.spark.SparkEnv$.create(SparkEnv.scala:339) at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:194) at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:279) at org.apache.spark.SparkContext.<init>(SparkContext.scala:464) at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2714) 请帮我解决这个问题。另外，我不明白为什么控件会进入“org.apache.spark.storage.StorageUtils$.(StorageUtils.scala:207)”的“if”块，因为该应用程序仅使用java 8编译。 问题不是由于代码进入 if 块造成的。这是因为它无法评估 if 语句。 NoSuchFieldError 大多数情况下是由于混合了同一库的不同版本而引起的。 在这种情况下，它与运行时使用的 org.apache.commons.lang3.JavaVersion 不具有 Java_9 属性有关。可能是因为您正在拉取太旧版本的 Apache Commons Lang 3 库，而 Spark 需要更新版本。 您可以运行mvn dependency:tree来检查直接或传递拉取的版本是否存在不一致。

Databricks 笔记本内核在运行 Pandas 聚合时崩溃

我的问题更具概念性，与 Databricks 如何分配计算资源有关。 我遇到一个问题，当我尝试在一个小数据集上运行一系列 pandas 聚合时（...

这种情况应该建立哪种类型的索引来加快Hudi表的查询速度？

我有一个Spark生成的Hudi表；架构如下： id：int64 内容：字符串 创建日期：时间戳[ns] 这张桌子超级大。我们在此表上执行的大多数查询都涉及...

如何计算Dataframe大小？

在Pyspark中，如何通过Pyspark中的任何可用方法查找数据帧大小。？我的生产系统正在运行 < 3.0 spark version. The idea is based on dataframe size i need to calculate

Spark 结构体到 getAs[T] 的类转换问题

我经常在 Spark 数据集行上使用映射函数来在 Scala 中对类型化对象进行转换。我通常的模式是转换从数据帧转换创建的中间结果（使用......

如何在 fs.azure.account.oauth2.msi.endpoint 中动态设置“api-version”

目前我正在通过 pyspark 库使用 hadoop-azure-3.4.1 连接到 ABFS。根据文档 - https://hadoop.apache.org/docs/stable/hadoop-azure/abfs.html#Azure_Managed_Identity -...

如何使用pyspark启动独立集群？

我在ubuntu下使用pyspark和python 2.7 我安装它使用 pip 安装 pyspark --用户 并尝试按照说明设置 Spark 集群 我找不到脚本启动大师...

描述表查看哪些列不为空

在 Apache Spark SQL (Azure Databricks) 中，我创建了一个如下表： 创建表 t( 一个大整数， b BIGINT 不为空 ） 我已验证我有一个可为空的列和一个不可为空的列...

生成列排列的最佳方法是什么

我有一个 scala DF，如下所示： +---+----+----+----+----+ |ID |信息 |col1|col2|col3| +---+----+----+----+----+ |id1|info1|a1 |a2 |a3 | |id2|info2|a1 |a3 |a4 | +---+-----+----+----+-...

使用 Spark 分解大型 JDBC 写入

我们想要将大型 Spark 数据帧复制到 Oracle 中，但我发现调整选项有点有限。查看 Spark 文档，这是我能找到的 JDBC 令状的唯一相关调整属性...

利用大 Parquet 数据和高磁盘使用率优化 PySpark 作业

我目前正在优化一个 PySpark 作业，该作业涉及跨大型数据集的一些聚合。我对处理大规模数据相当陌生，并且遇到了磁盘问题...

并行线程池执行器

我有 PySpark 代码，它很少对外部系统进行 POST API 调用。 对于输入数据帧中的每一行，我需要触发 POST API 请求（使用 Python 代码）以在外部创建一个条目...

收到“无法广播大于 8GB 的表错误” - 澄清

这更多的是对我的理解的澄清。我收到错误“无法广播大于 8GB 的表错误” 这是我的伪代码： val RiskDF = 广播(someDF) // 我...