问题标签 [spark2.4.4]

我尝试复制此Spark / Scala 示例，但是当我尝试从已处理的 .csv 文件中提取一些指标时出现错误。

我的代码片段：

当我尝试了解该属性时areaUnderPR，出现此错误：

20/01/10 10:41:02 WARN TaskSetManager：在阶段 56.0 中丢失任务 0.0（TID 246、10.10.252.172、执行程序 1）：java.lang.ClassNotFoundException：在 java.net.URLClassLoader 的预测.TestCancerOriginal$$anonfun$1 .findClass(URLClassLoader.java:382) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:424) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:357) 在 java.lang.Class.forName0(Native Method ) 在 java.lang.Class.forName(Class.java:348) 在 org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:67) 在 java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream. java:1868) 在 java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1751) 在 java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2042) 在 java.io.ObjectInputStream。readObject0(ObjectInputStream.java:1573) 在 java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287) 在 java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211) 在 java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java :2069) 在 java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573) 在 java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287) 在 java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211) 在 java .io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069) 在 java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573) 在 java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287) 在 java.io.ObjectInputStream。在 java.io 上读取串行数据（ObjectInputStream.java:2211）。ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431) at org.apache.spark.serializer.JavaDeserializationStream .readObject(JavaSerializer.scala:75) 在 org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:114) 在 org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:88) 在 org. apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at org.apache.spark.scheduler.Task.run(Task.scala:123) at org.apache.spark.executor.Executor$TaskRunner$$anonfun $10.apply(Executor.scala:408) 在 org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) 在 org.apache.spark.executor.Executor$TaskRunner。在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 在 java.lang.Thread 的 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 运行(Executor.scala:414) .run(Thread.java:748)

我的 predictions.show 结果：

我正在使用 Pyspark 2.4.4，我需要使用 UDF 来创建我想要的输出。此 UDF 使用广播字典。首先，看起来我需要修改 UDF 的代码以接受字典。其次，我不确定我正在做的事情是在 Spark 2.4 中最有效的方式。我的代码如下：

我的第一个问题是关于字典的。我应该使用它还是有其他更有效的方法？我正在考虑 collectAsMap()，但鉴于它可用于 rdds，我不确定这是否是 Spark 2.4 中的方式。

第二个问题是，鉴于字典是要走的路，我应该如何修改 udf 函数？

提前致谢！

I need to output the Spark application name (spark.app.name) in each line of the driver log (along with other attributes like message and date). So far I failed to find the correct log4j configuration or any other hints. How could it be done?

I would appreciate any help.

Using Spark standalone mode.

当两个或多个 Spark 作业具有相同的输出目录时，文件的相互删除将是不可避免的。

我正在使用 spark 2.4.4 以附加模式编写数据帧，我想在 spark 的 tmp 目录中添加时间戳以避免这些删除。

例子：

我的 JobSpark 写在hdfs:/outputFile/0/tmp/file1.parquet

用其他数据调用相同的火花作业并写入hdfs:/outputFil/0/tm/file2.parquet

我希望 jobSpark1 写入，hdfs:/outputFile/0/tmp+(timeStamp)/file1.parquet 其他作业写入，hdfs:/outputFile/0/tmp+(timeStamp)/file2.parquet然后将镶木地板移动到 hdfs:/outputFile/

我们的数据每天加载到具有分区列的 hdfs 中。问题是每个分区的文件大小小于 50mb。因此，当我们从所有这些分区读取数据以将数据加载到下一张表时需要几个小时。我们如何解决这个问题？

环境

我通过 python API 使用 spark v2.4.4

问题

根据 spark文档，我可以通过设置以下配置强制 spark 下载所有 hive jar 以与我的 hive_metastore 交互

• spark.sql.hive.metastore.version=${my_version}
• spark.sql.hive.metastore.jars=maven

但是，当我运行以下 python 代码时，没有从 maven 下载任何 jar 文件。

我怎么知道没有下载jar文件？

1. log4j.logger.org.apache.spark.api.python.PythonGatewayServer=INFO通过在 $SPARK_HOME/conf/log4j.properties 中设置，我已将 logLevel=INFO 配置为默认值。我看不到任何日志表明 spark 正在与 maven 交互。根据这个我应该看到一个信息级别的日志
2. 即使由于某种原因我的日志记录被破坏了，SparkSession 对象也只是构建得太快而无法从 Maven 中拉出大罐子。它会在 5 秒内返回。如果我手动将 hive_metastore 的 maven 坐标添加到“spark.jars.packages”，则需要几分钟才能全部下载
3. 我已删除 ~/.ivy2 和 ~/.m2 目录以删除以前下载的缓存

其他测试

• 我还在 spark 3.0.0 集群上尝试过相同的代码，但它也不起作用
• 谁能发现我做错了什么？或者这个选项刚刚坏了

我正在尝试使用 insert overwrite 语句将数据插入表中，但出现以下错误。

命令如下

我正在尝试使用临时表，存储结果，然后在最终表中更新，但这也不起作用。

另外我正在尝试使用一些变量将记录插入表中，但这也不起作用。例如

请建议

我在我的 Windows 中安装了 Spark 2.4。这是我的生产环境所必需的。使用 Spark2.4 。现在，我还想测试 Spark3.0 功能。那么我可以在同一台 Windows 机器上安装 Spark-3.0 二进制文件而不会干扰 Spark-2.4 的安装吗？我不想使用 Linux 子系统。或虚拟机。

我已经使用 Spark2.4 在 scala 中编写了 UDAF。由于我们的 Databricks 集群处于不再支持的 6.4 运行时，我们需要迁移到具有长期支持并使用 Spark3 的 7.3 LTS。UDAF 在 Spark3 中已弃用，将来会被删除（很可能）。所以我正在尝试将 UDAF 转换为聚合器函数

使用后，输出为：

{“id”：1282，“名称”：“麦考密克圣诞节”}

{“id”：1305，“名称”：“麦考密克完美捏”}

{"id": 1677, "name": "Viking Cruises Viking Cruises"}

我想在我的 spark 工作中添加一个安全措施，如果他们在 X 小时后没有完成就杀死他们自己。（在yarn模式下在集群模式下使用spark 2.4.3）

在 spark 中没有找到任何可以帮助我实现我想要的配置。

我试着这样做：

但似乎并没有完成杀死应用程序，希望对此有任何想法或想法。

试图环顾四周，但没有找到任何好主意。