我使用 pySpark 编写镶木地板文件。我想更改该文件的 hdfs 块大小。我这样设置块大小，但它不起作用：

是否必须在开始 pySpark 作业之前设置？如果是这样，该怎么做。

我有一个 Spark Streaming 作业在我们的集群上与其他作业（Spark 核心作业）一起运行。我想对这些作业使用动态资源分配，包括 Spark Streaming。根据下面的 JIRA 问题，Spark Streaming（在 1.6.1 版本中）不支持动态分配。但在 2.0.0 中已修复

JIRA 链接

根据这个问题的PDF，它说应该有一个名为 spark.streaming.dynamicAllocation.enabled=true 但我在文档中没有看到这个配置的配置字段。

任何人都可以请确认，

1. 我不能在 1.6.1 版本中为 Spark Streaming 启用动态资源分配。
2. 它在 Spark 2.0.0 中可用吗？如果是，应该设置什么配置（spark.streaming.dynamicAllocation.enabled=true或spark.dynamicAllocation.enabled=true）

我想以 Kafka 作为数据源运行 Spark Streaming 应用程序。它在本地工作正常，但在集群中失败。我正在使用 spark 1.6.2 和 Scala 2.10.6。

这是源代码和堆栈跟踪。

DevMain.scala

对象 DevMain 使用 Logging { 扩展 App

我NullPointerException在第 6 行得到一个并且代码没有输入lme.parser。

这是lme.parser：

我已经登录 line1lme.parser并且它没有被打印并且它没有进入lem.parser。

... 3 更多

这是广播变量更改后的新异常

为了处理我拥有的数据，我之前提取了模式，因此当我读取数据集时，我提供了模式，而不是通过推断模式的昂贵步骤。

为了构造模式，我需要将几个不同的模式合并到最终的模式中，所以我一直在使用union (++)anddistinct方法，但我一直收到org.apache.spark.sql.AnalysisException: Duplicate column(s)异常。

例如，假设我们在以下结构中有两个模式：

输出：

我知道只有与另一个模式完全匹配的模式结构才会被distinct. 但是我希望结果看起来像这样：

其中所有的都被“组合”成一个模式。我已经筛选了scala 文档中的所有方法，但似乎找不到解决此问题的正确方法。有任何想法吗？

编辑：

最终目标是使用方法final_schema输入sqlContext.read.schema和读取 JSON 字符串的 RDD 。read

我正在使用spark version 1.6.3并yarn version 2.7.1.2.3附带HDP-2.3.0.0-2557. 因为spark版本在我使用的HDP版本中太旧了，我更喜欢远程使用另一个spark作为yarn模式。

这是我运行 spark shell 的方法；

一切似乎都很好，sparkContext正在初始化，sqlContext正在初始化。我什至可以访问我的蜂巢表。但是在某些情况下，当它尝试连接到块管理器时会遇到麻烦。

我不是专家，但我认为，当我在纱线模式下运行它时，块管理器正在我的纱线集群上运行。第一次觉得是网络问题，不想在这里问。但是，这发生在某些我还无法弄清楚的情况下。所以这让我觉得这可能不是网络问题。

这是代码；

下面的代码工作正常；

但是大小超过10，我不知道，每次运行都会改变；

这引发了一个异常；

我可能刚刚意识到，当有多个任务要洗牌时，就会发生这种情况。

问题是什么，是性能问题还是我看不到的其他网络问题。那是什么洗牌？如果是网络问题，是我的 spark 和 yarn 之间的问题，还是纱线本身的问题？

谢谢你。

编辑：

我只是在日志中看到了一些东西；

有时，在另一个块管理器上重试它是可行的，但是，因为超过了默认的最大允许次数 4，所以它大部分时间都不会结束。

编辑2：

Yarn 对此真的很沉默，但我认为这是网络问题，我可以将问题迭代到某个地方；

此 spark 部署在 HDP 环境之外。当 spark 向 yarn 提交应用程序时，yarn 会通知 spark 驱动程序有关块管理器和执行器的信息。Executors 是 HDP 集群中的数据节点，在其私有网络中具有不同的 IP。但是，当涉及到在集群外部通知 spark 驱动程序时，它为所有执行程序提供相同且始终单一的 IP。这是因为 HDP 集群中的所有节点都通过路由器并具有相同的 IP。假设 IP 是150.150.150.150，当 spark 驱动程序需要连接并向执行程序询问某些内容时，它会使用此 IP 进行尝试。但是这个IP实际上是整个集群的外部IP地址，而不是单个数据节点的IP。

有没有办法让纱线通过其私有 ip 通知执行者（块管理器）。因为，他们的私有 IP 也可以从这个 spark 驱动程序正在处理的机器上访问。

我想在 Scala 中使用窗口函数。

我有一个 CSV 文件，它是以下文件：

当我尝试在此数据框上应用窗口函数时，有时会起作用，有时会失败：

所以它有效！但是当我尝试使用更大的数字（包含上一个示例的行）时，我得到以下结果

我在 Spark 1.x 上，并试图读取 csv 文件。如果我需要指定一些数据类型，根据文档，我需要导入包org.apache.spark.sql.types中定义的类型。

当我在 spark-shell 中以交互方式使用它时，这很好用，但是因为我想通过 spark-submit 运行它，所以我编写了一些 Scala 代码来执行此操作。但是，当我尝试编译我的 Scala 代码时，它给了我一个错误，说它找不到 org.apache.spark.sql.types。我查找了 jar 的内容spark-sql，但找不到其中定义的这些类型。

那么，哪个 jar 有 org.apache.spark.sql.types？

我有一个 Spark 作业，它正在生成一组带有统计信息的结果。我的工作项目数量超过了奴隶数量。所以我对每个奴隶做不止一个处理。

因为我有多个写操作，所以我cache在生成RDD对象后能够重用它们：一个用于结果对象，另一个用于统计。两种写操作都使用saveAsHadoopFile.

在没有缓存的情况下，Spark 会在每次写入操作时再次重新运行该作业，这需要很长时间并重做两次相同的执行（如果我有更多的写入，则更多）。

通过缓存，我达到了内存限制。一些先前计算的结果在缓存期间丢失，我看到了"CacheManager:58 - Partition rdd_1_0 not found, computing it"消息。Spark 最终会进入一个无限循环，因为它试图缓存更多结果而丢失一些其他结果。

我知道 Spark 有不同的缓存存储级别。使用内存+磁盘可以解决我们的问题。但我想知道我们是否可以直接在工作人员中写下文件而不生成RDD对象。我不确定这是否可能。是吗？

从Spark 下载页面，如果我下载v2.0.1 的 tar 文件，我会看到它包含一些我认为可以包含在我的应用程序中的 jar。

如果我改为下载v1.6.2 的 tar 文件，则在其中找不到 jars 文件夹。是否有我应该从该站点使用的替代包类型？我目前正在选择默认值（为 Hadoop 2.6 预先构建）。或者，我可以在哪里找到这些 Spark jar - 我应该从http://spark-packages.org单独获取它们吗？

这是我想使用的一组指示性罐子：

• hadoop-通用
• 火花芯
• 火花-csv
• 火花-sql
• 单义解析器
• 火花催化剂
• json4s-核心

我在 Spark v1.6.0 上使用 spark.ml.util.SchemaUtils 时遇到问题。我收到以下错误：

在我的集群上运行这个最小示例时（受我最终想要使用的库的启发）：

但是，在我的桌面上本地启动的相同示例运行没有问题。

从我在网上看到的（例如这里），这种错误消息通常与编译和运行时环境之间的版本不匹配有关，但是我的程序、我的本地 spark 分发和我的集群分发都有相同的 Spark 和 mllib 版本v1.6.0，相同的 Scala 版本 v2.10.6，和相同的 Java 版本 v7。

我检查了 Spark 1.6.0 源代码，并且在 org.apache.spark.ml.util.SchemaUtils 中确实存在 appendColumn，并且具有正确的签名（但是 org.apache.spark.ml.util API 文档中没有提到 SchemaUtils ）。

ETA：从我的 pom.xml 文件中提取：