问题标签 [spark3]

我目前正在尝试将我的应用程序升级到 Spark 3.0.1。对于表创建，我使用 Python-Cassandra 连接器 cassandra-driver 删除并创建一个表。然后我使用 spark-cassandra 连接器将数据框写入表中。仅使用 spark-cassandra 连接器来创建和删除表并没有真正好的替代方法。

在 Spark 2.4 中，drop-create-write 流程没有问题。但是在 Spark 3.0 中，应用程序似乎没有特定的顺序来做这些事情，通常在删除和创建之前尝试编写。我不知道如何确保首先删除和创建表。我知道即使应用程序在写入时出错，也会发生删除和创建，因为当我通过 cqlsh 查询 Cassandra 时，我可以看到表被删除并重新创建。关于 Spark 3.0 中这种行为的任何想法？

注意：因为架构发生了变化，所以需要删除并重新创建这个特定的表，而不是直接覆盖。

根据要求的代码片段：

我正在 Kubernetes 上运行 Spark 3.0 应用程序（Spark Structured Streaming），并且正在尝试使用新的本机 Prometheus 度量接收器。我能够使其工作并获得此处描述的所有指标。

但是，我真正需要的指标是在启用以下配置时提供的指标：spark.sql.streaming.metricsEnabled，正如Spark Summit 演示文稿中所建议的那样。现在，即使将该配置设置为“true”，我也看不到/metrics/executors/prometheus广告中的任何流媒体指标。需要注意的一件事是，我可以在 下看到它们metrics/json，因此，我们知道配置已正确应用。

为什么不将流式指标发送到 Prometheus 接收器？我需要添加一些额外的配置吗？还不支持吗？

我最近从 2.4.6 升级到 Spark 3.0.1（以及 scala 2.11.12 到 scala 2.12.10）。我从 IntelliJ Idea 编写和执行应用程序，过去能够在将 Master 设置为本地 [*] 或使用 spark://xx:7077 远程运行的情况下运行。我的典型远程运行方法是提交由 IntelliJ 中可用的人工制品功能生成的 fat jar。

自升级以来，由于某种原因无法远程运行。这可能是什么原因？

SparkContext 设置

标准错误输出

任何解决此问题的帮助将不胜感激。

我正在将我的 Spark 版本从 2.4.5 升级到 3.0.1，我无法再加载使用“DecisionTreeClassifier”阶段的 PipelineModel 对象。

在我的代码中，我加载了几个 PipelineModel，所有带有阶段 ["CountVectorizer_[uid]"、"LinearSVC_[uid]"] 的 PipelineModel 都加载正常，而带有阶段 ["CountVectorizer_[uid]","DecisionTreeClassifier_[uid]" 的模型] 抛出以下异常：

AnalysisException：无法解析“ rawCount”给定的输入列：[gain、id、impurity、impurityStats、leftChild、prediction、rightChild、split]

这是我正在使用的代码和完整的堆栈跟踪：

这些管道模型使用 Spark 2.4.3 保存，我可以使用 Spark 2.4.5 很好地加载它们。

我试图进一步调查并分别加载每个阶段。加载 CountVectorizerModel

产生一个 CountVectorizerModel，这样就可以了，但是我的代码在尝试加载 DecisionTreeClassificationModel 时失败：

这是我的决策树分类器的“数据”的内容：

我正在使用Spark 3.0.1和用户提供的Hadoop 3.2.0和Scala 2.12.10在Kubernetes上运行。

读取压缩为snappy的 parquet 文件时一切正常，但是当我尝试读取压缩为 zstd 的 parquet 文件时，几个任务在以下错误下失败：

我没有得到的是，这些任务在重试后会成功，但并非总是如此，因此我的工作经常失败。如前所述，如果我使用与 snappy 相同的数据集压缩，一切正常。

我还尝试构建 Spark 和 Hadoop，更改 zstd-jni 版本，但仍然会发生相同的行为。

有谁知道可能会发生什么？

谢谢！

我编写了一个自定义Aggregator（ 的扩展org.apache.spark.sql.expressions.Aggregator），Spark 将其作为语句下的聚合函数正确调用group by：

不过，我想在窗口函数中使用它，因为排序对我很重要。我试过这样调用它：

这就是我得到的错误：

是否可以在 Spark 3.0.1 中使用自定义聚合器作为窗口函数？如果是这样，我在这里错过了什么？

Currently I am studying the usage of Apache Spark 3.0 with Rapids GPU Acceleration. In the official spark-rapids docs I came across this page which states:

There are cases where you may want to get access to the raw data on the GPU, preferably without copying it. One use case for this is exporting the data to an ML framework after doing feature extraction.

To me this sounds as if one could make data that is already available on the GPU from some upstream Spark ETL process directly available to a framework such as Tensorflow or PyTorch. If this is the case how can I access the data from within any of these frameworks? If I am misunderstanding something here, what is the quote exactly referring to?

I've tried to use Spark AQE for dynamically coalescing shuffle partitions before writing. On default, spark creates too many files with small sizes. However, AQE feature claims that enabling it will optimize this and merge small files into bigger ones. This is critical for aws s3 users like me because having too many small files causes network congestion when trying to read the small files later.

Here is my spark configuration:

The required parameters for AQE are all enabled, I also see AdaptiveSparkPlan isFinalPlan=true in the execution plan. When I run a small task (read a csv, do some calculations, do a join operation and write into parquet), it still generates too many small sized files in the parquet folder. Am i missing something or this feature is not doing what it promised?

我有以下json：

我正在尝试这样阅读：

但它无法在数据框中正确显示我的记录，我如何绕过那个“值”嵌套以正确地将我的行放在数据框中？

当前结果：

我试图得到的结果是：

想法中

版本：spark3.0.1，scala2.12.12，java1.8.0_212

我的代码：

错误：

如果使用 spark2.4.7 和 scala2.11 就可以了

所以有什么问题？