问题标签 [pyspark]

我在 Spark 1.3.0（和 1.4.0）中遇到了 cache() 数据框的问题。

根据文档： http ://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.HiveContext

不管我是否使用以下任何方法： cache() -- 保持默认存储级别 (MEMORY_ONLY_SER)。cacheTable(tableName) -- 在内存中缓存指定的表。

该方法运行时不会被炸毁，但是当我在 Spark 中对我的框架或表执行后续操作时，它会给我这个错误消息。

谢谢你的帮助！

看似简单的问题，却找不到答案。

问题：我创建了一个函数，我将把它传递给 map()，它接受一个字段并从中创建三个字段。我希望 map() 的输出给我一个新的 RDD，包括来自输入 RDD 和新/输出 RDD 的字段。我该怎么做呢？

我是否需要将我的数据的键添加到函数的输出中，以便我可以将更多的输出 RDD 加入到我的原始 RDD 中？这是正确/最佳做法吗？

更基本的是，我似乎无法将两个 Row 组合起来。

这不会像我想要的那样返回 new Row() 。

谢谢

假设我有一个数据集

我要做的是将值平面映射到作为列索引的键和作为值的值。谁能给我指导？我发现很难获得列索引。

我想利用 python 的极其简单的文本解析和函数式编程能力，也想利用 numpy 和 scipy 等科学计算库的丰富产品，因此我想使用 pyspark 来完成一项任务。

我一开始希望执行的任务是从存储桶中读取，其中有文本文件作为流的一部分写入。有人可以粘贴如何使用 pyspark 从 s3 路径读取流数据的代码片段吗？直到最近，我认为这只能使用 scala 和 java 来完成，但我今天才发现 spark 1.2 以后，pyspark 也支持流式传输，但不确定是否支持 S3 流式传输？

我以前在 scala 中执行此操作的方式是将其作为 HadoopTextFile 读取，并且还使用配置参数来设置 aws 密钥和秘密。我将如何在 pyspark 中做类似的事情？

任何帮助将非常感激。

提前致谢。

我有Python应该在Apache Spark集群上运行的基于 - 的脚本。

我有Hadoop MapReduce InputFormat作为RDD. 这里没有问题。

问题是我想Hadoop Configuration用加载的附加资源文件和属性集来构建自定义。意图是在Configuration里面使用修饰符Python SparkContext。

我可以构建JVM可以构建和加载所需的代码Hadoop Configuration。如何将其附加到Python使用中PySpark？

有谁知道这一切是如何实现的？

有两个id: bigint，我想删除一个。我能怎么做？

我写了一个小脚本，有两个函数，一个生成随机日志并提供给 kafka 生产者，另一个使用 kafka 主题在 Spark Streaming 中创建数据流。

我希望这两个函数使用 python 多处理同时工作，不幸的是，当我运行我的脚本时，我得到一个与 KafkaUtils.createStream 相关的错误......

这是我的终端显示的内容：

我的两个功能：

谢谢你的帮助 ！

铅：

a) 将本地文件读入 Panda 数据帧，例如 PD_DF b) 操作/Massge PD_DF 并将列添加到数据帧 c) 需要使用 spark 将 PD_DF 写入 HDFS。我该怎么做 ？

我在本地窗口上使用独立集群，并尝试使用以下代码从我们的一台服务器加载数据 -

我已将 SPARK_CLASSPATH 设置为 -

执行 sqlContext.load 时，它会抛出错误，提示“找不到适合 jdbc:postgresql 的驱动程序”。我已经尝试搜索网络，但无法找到解决方案。

我正在尝试使用测试机器学习数据集在 pyspark 中运行 Spark MLlib 包。我将数据集分成一半的训练数据集和一半的测试数据集。下面是我构建模型的代码。但是，它显示了所有因变量的 NaN、NaN.. 的权重。想不通为什么。但是当我尝试使用 StandardScaler 函数标准化数据时，它会起作用。

非常感谢你的帮助。

下面是我用来进行缩放的代码。