从 Pyspark UDF 调用另一个自定义 Python 函数
假设您有一个文件，我们将其命名为 udfs.py 并在其中: defnested_f(x):返回 x + 1def main_f(x):返回nested_f(x) + 1 然后您想从 main_f 函数中创建一个 UDF 并在数据帧上运行它: import pyspark.sql.functions as fn将熊猫导入为 pdpdf = pd.DataFrame([[1], [2], [3
..
pyspark相关内容
ModuleNotFoundError 因为 PySpark 序列化程序无法找到库文件夹
我有以下文件夹结构 - 库文件夹- lib1.py- lib2.py- main.py main.py 调用 libfolder.lib1.py 然后调用 libfolder.lib2.py 和其他. 在本地机器上一切正常，但在我将其部署到 Dataproc 后，出现以下错误 文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/seria
..
如何将文件传递给主节点?
我已经用python编写了代码来实现二进制分类，我想使用Apache-Spark根据本地计算机中的不同数据文件并行化这个分类过程.我已经完成了以下步骤: 我编写了包含 4 个 python 文件的整个项目:“run_classifer.py"(用于运行我的分类应用程序)、“classifer.py"(用于二进制分类)、“load_params.py"(用于加载用于分类的学习参数)和“prep
..
Pyspark RDD ReduceByKey 多功能
我有一个名为 DF 的 PySpark DataFrame，带有 (K,V) 对.我想使用 ReduceByKey 应用多个函数.例如，我有以下三个简单的功能: def sumFunc(a,b):返回a+bdef maxFunc(a,b): 返回 max(a,b)def minFunc(a,b): 返回 min(a,b) 当我只应用一个函数时，例如，以下三项工作: DF.reduceByKe
..
PySpark“爆炸"列中的字典
我在 spark 数据框中有一列“true_recoms": -RECORD 17-----------------------------------------------------------------项目 |20380109true_recoms |{"5556867":1,"5801144":5,"7397596":21} 我需要“爆炸"这个列才能得到这样的东西: item |
..
如何使用 PySpark 转换结构化流?
这看起来应该是显而易见的，但是在查看文档和示例时，我不确定我是否可以找到一种使用 PySpark 获取结构化流和转换的方法. 例如: from pyspark.sql import SparkSession火花=(火花会话.builder.appName('StreamingWordCount').getOrCreate())原始记录 = (火花.readStream.format('so
..
Spark 不再在 Windows 中运行
我有 Windows 10 并且我遵循了本指南安装 Spark 并使其在我的操作系统上运行，只要使用 Jupyter Notebook 工具即可.我使用这个命令来实例化 master 并导入我工作所需的包:pyspark --packages graphframes:graphframes:0.8.1-spark3.0-s_2.12 --master local[2] 然而，后来，我发现没有
..
如何选择最后一行以及如何通过索引访问 PySpark 数据框?
来自像 这样的 PySpark SQL 数据帧 name 年龄城市abc 20定义 30 B 如何获取最后一行.(就像通过 df.limit(1) 我可以将数据帧的第一行放入新的数据帧中). 以及如何通过 index.like 行号访问数据帧行.12 或 200 . 在熊猫中我可以做到 df.tail(1) # 最后一行df.ix[rowno or index] # 按索引df
..
PySpark — UnicodeEncodeError: 'ascii' 编解码器无法编码字符
使用 spark.read.csv 和 encoding='utf-8' 将带有外来字符 (åäö) 的数据帧加载到 Spark 中，并尝试做一个简单的展示(). >>>df.show()回溯(最近一次调用最后一次):文件“"，第 1 行，在 中文件“/usr/lib/spark/python/pyspark/sql/dataframe.py"，第287行，显示打
..
使用 PySpark 删除 Dataframe 的嵌套列
我正在尝试使用 pyspark 在 Spark 数据框中删除一些嵌套列.我发现这个 Scala 似乎正在做我想做的事，但我不熟悉 Scala，也不知道如何用 Python 编写它. https://stackoverflow.com/a/39943812/5706548 我非常感谢您的帮助. 谢谢， 解决方案 我发现使用 pyspark 的一种方法是首先将嵌套列转换为 j
..
未找到密钥:_PYSPARK_DRIVER_CALLBACK_HOST
我正在尝试运行此代码: 导入pyspark从 pyspark.sql 导入 SparkSessionspark = SparkSession.builder \.master("本地") \.appName("字数统计") \.getOrCreate()df = spark.createDataFrame([(1, 144.5, 5.9, 33, 'M'),(2, 167.2, 5.4, 45,
..
Pyspark - 将 json 字符串转换为 DataFrame
我有一个包含简单 json 的 test2.json 文件: { "Name": "something", "Url": "https://stackoverflow.com", "Author": "jangcy", "BlogEntries": 100, "Caller": "jangcy"} 我已将文件上传到 blob 存储并从中创建了一个 DataFrame: df = spark.
..
pyspark 列不可迭代
当我尝试 groupBy 并获得 max 时，有了这个数据框，我得到的 Column 是不可迭代的: linesWithSparkDF+---+-----+|编号|周期|+---+-----+|31|26||31|28||31|29||31|97||31|98||31|100||31|101||31|111||31|112||31|113|+---+-----+只显示前 10 行ipython-
..
PySpark:具有不同列的 DataFrame 的动态联合
考虑此处显示的数组.我有 3 组数组: 数组 1: C1 C2 C31 2 39 5 6 数组 2: C2 C3 C411 12 1310 15 16 数组 3: C1 C4111 112110 115 我需要如下输出，输入我可以获得 C1、...、C4 的任何一个值，但在加入时我需要获得正确的值，如果该值不存在，则它应该为零. 预期输出: C1 C2 C3 C41 2 3
..
PySpark:彻底清理检查点
根据文档可以告诉 Spark 跟踪 “超出范围" 检查点 - 那些不再需要的 - 并从磁盘中清除它们. SparkSession.builder....config("spark.cleaner.referenceTracking.cleanCheckpoints", "true").getOrCreate() 显然是这样做的，但问题是最后一个检查点的 rdd 永远不会被删除. 问题
..
如何将列声明为 DataFrame 中的分类特征以用于 ml
如何声明我的 DataFrame 中的给定列包含分类信息? 我有一个从数据库加载的 Spark SQL DataFrame.这个 DataFrame 中的许多列都有分类信息，但它们被编码为 Longs(为了隐私). 我希望能够告诉 spark-ml，即使此列是数值，但信息实际上是分类的.类别的索引可能有一些漏洞，这是可以接受的.(例如，一列可能有值 [1, 0, 0 ,4]) 我
..
如何为 Scala Spark ETL 设置本地开发环境以在 AWS Glue 中运行?
我希望能够在我的本地 IDE 中编写 Scala，然后将其部署到 AWS Glue 作为构建过程的一部分.但我无法找到构建 AWS 生成的 GlueApp 框架所需的库. aws-java-sdk-glue 没有't 包含导入的类，我在其他任何地方都找不到这些库.虽然它们必须存在于某个地方，但也许它们只是这个库的 Java/Scala 端口:aws-glue-libs 来自 AWS 的
..
在来自 google-dataproc 的 Spark 集群中的 pyspark 作业中使用外部库
我有一个通过 google dataproc 创建的 Spark 集群.我希望能够使用 databricks 中的 csv 库(参见 https://github.com/databricks/spark-csv).所以我先是这样测试的: 我与集群的主节点启动了 ssh 会话，然后输入: pyspark --packages com.databricks:spark-csv_2.11:1.
..
数据帧与 Apache Spark 中的 pyspark 转置
我有一个具有以下结构的数据帧 df: +-----+-----+-----+-------+|s |col_1|col_2|col_...|+-----+-----+-----+-------+|f1 |0.0|0.6|... ||f2 |0.6|0.7|... ||f3 |0.5|0.9|... ||...|...|...|... | 我想计算这个数据帧的转置，所以它看起来像 +-----
..
仅从 pyspark 中的 Spark DF 中选择数字/字符串列名称
我在 pyspark (2.1.0) 中有一个 SparkDataFrame，我希望仅获取数字列的名称或仅获取字符串列的名称. 例如，这是我的 DF 的架构: root|-- 性别:字符串(可为空 = 真)|-- SeniorCitizen: string (nullable = true)|-- MonthlyCharges: double (nullable = true)|-- To
..