问题标签 [apache-spark-2.0]

我正在使用 pyspark 2.0 通过读取 csv 来创建 DataFrame 对象：

我找到使用的数据类型

结果是

我正在尝试将数据中的某些列转换为 LabeledPoint 以应用分类。

我遇到了这个问题：

对错误有任何想法吗？有没有办法从 DataFrame 生成 LabelPoint 以执行分类？

我有一个包含如下数据的数据集：

...

现在，我想将数据分组如下（col1：String Key，col2：List）：

我认为使用 goupByKey 将是一个足够的解决方案，但我找不到任何示例，如何使用它。

任何人都可以帮助我找到使用 groupByKey 或使用任何其他转换和操作的组合来通过使用数据集而不是 RDD 来获得此输出的解决方案吗？

我想调整我的 Java Spark 应用程序（它实际上使用 RDD 进行某些计算）以使用Datasets而不是RDDs. 我是数据集的新手，不知道如何将哪个事务映射到相应的数据集操作。

目前我像这样映射它们：

相应的问题是：

• 等于JavaRDD.mapToPair方法Dataset.groupByKey？
• JavaPairRDD映射KeyValueGroupedDataset到?
• 哪种方法等于JavaPairRDD.aggregateByKey方法？

但是，我想将以下 RDD 代码移植到数据集中：

我的代码看起来是这样的：

我正在使用火花 2.0.0。有没有办法将参数从火花驱动程序传递给执行程序？我尝试了以下。

如上所述，我编写了一个自定义 mappartitionfunction 来传递参数。并在 partitionfunction 的调用方法中访问静态变量。这在我使用“setmaster（”local”）在本地运行时有效。但在使用 .setmaster（“yarn”）的集群上运行时不起作用。（在 system.out.println 语句中打印为 null）

有没有办法将参数从驱动程序传递给执行程序。

我正在使用 Spark 2.0，这项工作首先对输入数据进行排序并将其输出存储在 HDFS 上。

我遇到了内存错误，解决方案是将“spark.shuffle.memoryFraction”的值从 0.2 增加到 0.8，这解决了问题。但是在文档中我发现这是一个不推荐使用的参数。

据我了解，它已被“spark.memory.fraction”取代。如何在考虑到 HDFS 上的排序和存储的同时修改此参数？

我在 S3 上有一些使用 snappy 压缩算法（使用node-snappy包）压缩的 csv 文件。我喜欢使用 spark 处理这些文件，com.databricks.spark.csv但我一直收到无效的文件输入错误。

代码：

错误信息：

16/09/24 21:57:25 WARN TaskSetManager：在阶段 0.0 中丢失任务 0.0（TID 0，ip-10-0-32-5.ec2.internal）：java.lang.InternalError：无法解压缩数据。输入无效。在 org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompress(SnappyDecompressor.java:239) 在 org.apache.hadoop 的 org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompressBytesDirect(Native Method)。 io.compress.BlockDecompressorStream.decompress(BlockDecompressorStream.java:88) at org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:85) at java.io.InputStream.read(InputStream.java:101)在 org.apache.hadoop.util.LineReader.fillBuffer(LineReader.java:180) 在 org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:216) 在 org.apache.hadoop.util.LineReader.readLine （LineReader.java：

我的 spark r 1.6 代码在 spark2.0 中不起作用，我进行了必要的更改，例如创建sparkr.session()而sparkr.init()不是传递 sqlcontext 参数等……</p>

在下面的代码中，我将几个文件夹中的数据加载到数据框中

spark1.6 中的 read.df 有效

spark2.0 中的 read.df 不起作用

上面的行引发以下错误：

我需要在 spark2 中一次写入两个输出。我试过这样的事情：

这将做我需要的，但它会执行整个链两次，这可能很耗时，我不需要做所有的转换两次我只需要写两次。

我有办法吗？

谢谢

我正在尝试从 spark-1.6 迁移到 spark-2.0，并尝试使用 spark Cassandra 连接器 2.0 进行一些基本数据操作

我只是从表中获取数据并使用 rdd.foreach(println) 打印它

但这给了我堆栈跟踪错误

对此有任何想法。

此外，cassandraSQLContext 也已从最新版本的 spark cassandra 连接器中删除，所以有什么替代方案吗？

我们正在尝试使用 ThriftServer 从 spark 临时表中查询数据，在 spark 2.0.0 中。

首先，我们创建了启用 Hive 支持的 sparkSession。目前，我们使用 sqlContext 启动 ThriftServer，如下所示：

我们有注册临时表“spark_temp_table”的火花流：

使用直线，我们可以看到临时表（运行 SHOW TABLES）；

当我们想用这种方法运行第二个作业（使用第二个 sparkSession）时，我们必须使用不同的端口启动第二个 ThriftServer。

我在这里有两个问题：

1. 有什么方法可以在一个端口上拥有一个 ThriftServer 以访问不同 sparkSessions 中的所有临时表？

2. HiveThriftServer2.startWithContext(spark.sqlContext());用 注释@DeveloperApi。有没有办法以编程方式在代码中没有上下文的情况下启动节俭服务器？
   我看到启动时有配置--conf spark.sql.hive.thriftServer.singleSession=true传递给 ThriftServer（sbin/start-thriftserver.sh），但我不明白如何为作业定义它。我尝试在 sparkSession builder 中设置此配置属性，但直线没有显示临时表。