问题标签 [spark-csv]

这就是我在 spark 数据框中加载 csv 文件的方式

这是我的输入文件名之一的示例。

现在我想读取这个文件并用“。”分割它。运算符，然后添加 CUS 作为新列代替 DataPartition 。

我可以在没有任何 UDF 的情况下做到这一点吗？

这是现有数据框的架构

建议答案后更新代码

我有一个数据框，我将其连接到它的所有字段。

连接后它成为另一个数据帧，最后我将其输出写入 csv 文件，并在其两列上进行了分区。它的一列存在于第一个数据框中，我不想将其包含在最终输出中。

这是我的代码：

在这里，我正在连接并创建另一个数据框：

这是我尝试过的

现在我不想在我的输出中出现 DataPartition 列。

我正在基于 DataPartition 进行分区，所以我没有得到，但因为 DataPartition 存在于主数据框中，所以我在输出中得到它。

问题 1： 如何忽略 Dataframe 中的列

问题2：在写入我的实际数据之前，有什么方法可以"\ufeff"在csv输出文件中添加，这样我的编码格式就会变成UTF-8-BOM。

根据建议的答案

这是我尝试过的

但低于错误

下面是我是否必须在最终输出中删除两列的问题

我有两个数据集，数据集 1 在下面

这就是我使用自动发现模式加载数据的方式

数据集 2：

我从两者中创建一个数据框，然后加入。最后，我将两个数据帧的输出写入 csv 文件。

这是写入 csv 文件的代码。

一切正常，除了.option("nullValue", ""). 我无法用空白值替换 null。

在我的输出中，我仍然看到空值。

我也试过这个，但得到了相同的结果。

我有一个看起来像这样的数据集

这就是我使用自动发现模式加载数据的方式

现在我有另一个数据框，我在其上进行连接操作，最后我创建了一个将输出写入 csv 文件的数据框。

最终数据框如下所示

现在在我的输出文件中，我只看到我的标题 concatenated是预期的。

现在我的问题是无论如何将我的最终输出的标题更改为df1result数据帧的标题

我必须在 spark 数据框中的每个分区的文件中进行记录计数，然后我必须将输出写入 XML 文件。

这是我的数据框。

现在我必须计算每个分区中每个文件的记录数，然后将输出写入 XML 文件。

这就是我试图做到的方式。

我能够打印每个分区的记录总数并打印，但是当我尝试创建 xml 文件时，我遇到了错误。

我在用 Spark 2.2.0, Zeppelin 0.7.2

所以我必须导入com.databricks.spark.xml这个，但为什么因为如果我不导入 .csv 文件的话com.databricks.spark.csv。

另外，我可以使用缓存dfMainOutputFinalWithoutNull，因为我将使用它两次来写入它的数据，然后计算它的分区记录，然后写入 xml 文件？

我添加了这个依赖

并重新启动解释器。然后我收到以下错误。

我正在对两个分别具有 280 GB 和 1 GB 数据的数据帧进行连接。我实际的计算连接的火花作业速度很快，但随机读取和写入需要很长时间，这使得整个火花作业非常慢。

我正在使用 m3.2xlarge 10 节点集群。

请建议我应该做哪些必要的更改以使这个随机读写阶段更快。

添加我的完整代码..

我正在尝试将所有火花输出部分文件合并到一个目录中，并在 Scala 中创建一个文件。

这是我的代码：

然后在最后一步，我正在编写如下所示的数据帧输出。

当我运行它时，我得到以下异常

这就是我获得输出的方式

而不是文件夹，我想合并文件夹内的所有文件并创建一个文件。

我有一个数据框，我在其中将默认分隔符替换,为|^|. 它工作正常，我也得到了预期的结果，除了在,记录中找到的位置。例如，我有一个这样的记录，如下所示

所以,在第四场。

现在我这样做是为了替换,

我在保存的输出部分文件中得到这样的输出

我的问题是" "在结果的开头和结尾。

如果删除逗号，那么我会得到正确的结果，如下所示

我正在尝试将数据框保存为本地驱动器中的 CSV 文件。但是，当我这样做时，我会生成一个文件夹，并在该分区中写入文件。有什么建议可以克服这个吗？

我的要求： 获取代码中给出的实际名称的普通 csv 文件。

代码片段： dataframe.coalesce(1).write.mode("overwrite").format("com.databricks.spark.csv").option("header", "true").csv("E:/dataframe.csv")

我正在尝试使用 Spark Streaming 从 NFS 上的 CSV 文件中收集数据。我的代码非常简单，到目前为止我只在 spark-shell 中运行它，但即使在那里我也遇到了一些问题。

我正在运行带有 6 个工作人员的独立 Spark 主机的 spark-shell，并将以下参数传递给 spark-shell：

--master spark://master.host:7077 --num-executors 3 --conf spark.cores.max=10

这是代码：

该 NFS 路径中有 2 个文件，每个文件大小约为 200MB。当我调用 writeStream 时，我收到以下警告：

“17/11/13 22:56:31 WARN TaskSetManager：第 2 阶段包含一个非常大的任务 (106402 KB)。建议的最大任务大小为 100 KB。”

查看 Spark 主 UI，我看到只使用了一个执行程序——创建了四个任务，每个任务读取每个 CSV 文件的约 50%。

我的问题是：

1) NFS 路径中的文件越多，驱动程序似乎需要的内存就越多 - 如果有 2 个文件，它会崩溃，直到我将其内存增加到 2g。4个文件需要不少于8g。驱动程序在做什么，它需要这么多内存？

2) 如何控制读取 CSV 文件的并行度？我注意到文件越多，创建的任务就越多，但是可以手动控制吗？