如何在 Pyspark 中替换数据帧的所有空值
我在 pyspark 中有一个超过 300 列的数据框.在这些列中有一些值为 null 的列. 例如: Column_1 column_2空空空空第234话125 124365 187等等 当我想对 column_1 求和时，结果是 Null，而不是 724. 现在我想用空白替换数据框所有列中的空值.因此，当我尝试对这些列求和时，我不会得到空值，但会得到一个数值. 我们如何在
..
pyspark相关内容
Web UI 如何计算存储内存(在 Executors 选项卡中)?
我正在尝试了解 Spark 2.1.0 如何在节点上分配内存. 假设我正在启动一个本地 PySpark REPL，为其分配 2GB 内存: $ pyspark --conf spark.driver.memory=2g Spark UI 告诉有 956.6 MB 分配给存储内存: 我不明白如何得到那个数字，这是我的思考过程: 驱动程序堆大小设置为2048 MB， 根据文档:
..
PySpark isin 函数
我正在使用 PySpark 将我的旧 Python 代码转换为 Spark. 我想要一个 PySpark 等价物: usersofinterest = actdataall[actdataall['ORDValue'].isin(orddata['ORDER_ID'].unique())]['User ID'] actdataall 和 orddata 都是 Spark 数据帧.
..
如何在 PySpark 中用 NULL 替换字符串值?
我想做这样的事情: df.replace('空值', None, 'NAME') 基本上，我想用 NULL 替换一些值.但它在这个函数中不接受 None .我该怎么做? 解决方案 这会将 name 列中的 empty-value 替换为 None: from pyspark.sql.functions import udf从 pyspark.sql.types 导入 StringTy
..
正确使用大型广播变量的技巧?
我使用了一个大约 100 MB 大小的腌制广播变量，我用它来近似: >>>数据 = 列表(范围(整数(10*1e6)))>>>导入 cPickle 作为泡菜>>>len(pickle.dumps(数据))98888896 在具有 3 个 c3.2xlarge 执行程序和一个 m3.large 驱动程序的集群上运行，使用以下命令启动交互式会话: IPYTHON=1 pyspark --ex
..
Spark 上下文“sc"未定义
我是 Spark 的新手，我正在尝试通过参考以下站点来安装 PySpark. http://ramhiser.com/2015/02/01/configuring-ipython-notebook-support-for-pyspark/ 我尝试安装预构建包，也尝试通过 SBT 构建 Spark 包. 当我尝试在 IPython Notebook 中运行 python 代码时，出
..
环境变量 PYSPARK_PYTHON 和 PYSPARK_DRIVER_PYTHON
我最近安装了pyspark.它已正确安装.当我在 python 中使用以下简单程序时，出现错误. >>from pyspark import SparkContext>>sc = SparkContext()>>数据 = 范围(1,1000)>>rdd = sc.parallelize(data)>>rdd.collect() 在运行最后一行时出现错误，其关键行似乎是 [阶段 0:>(0 +
..
如何在pyspark中分解数据框的多列
我有一个数据框，其中包含与以下类似的列中的列表.所有列中列表的长度都不相同. 姓名年龄科目成绩[鲍勃] [16] [数学、物理、化学] [A,B,C] 我想以得到以下输出的方式分解数据框- 姓名年龄科目成绩鲍勃 16 数学 A鲍勃 16 物理 B鲍勃 16 化学 C 我怎样才能做到这一点? 解决方案 这行得通， import pyspark.sql.functions as F从
..
当值与 pyspark 中的字符串的一部分匹配时过滤 df
我有一个很大的 pyspark.sql.dataframe.DataFrame，我想保留(所以 filter)URL 保存在 location 列包含一个预先确定的字符串，例如'google.com'. 我试过了: import pyspark.sql.functions as sfdf.filter(sf.col('location').contains('google.com')).s
..
Spark 1.4 增加 maxResultSize 内存
我正在使用 Spark 1.4 进行研究并在内存设置方面苦苦挣扎.我的机器有 16GB 的内存，所以没有问题，因为我的文件大小只有 300MB.虽然，当我尝试使用 toPandas() 函数将 Spark RDD 转换为熊猫数据帧时，我收到以下错误: 9 个任务的序列化结果 (1096.9 MB) 大于 spark.driver.maxResultSize (1024.0 MB) 我尝试通过
..
为什么我从 date_format() PySpark 函数得到空结果?
假设有一个日期框架，其中一列由日期作为字符串组成.对于该假设，我们创建以下 dataFrame 作为示例: # 导入 sql 类型从 pyspark.sql.types 导入 StringType、IntegerType、StructType、StructField、DoubleType、FloatType、DateType从 pyspark.sql.functions 导入 date_form
..
全局启用 spark.sql 区分大小写
选项 spark.sql.caseSensitive 控制列名等是否应该区分大小写.它可以设置，例如通过 spark_session.sql('set spark.sql.caseSensitive=true') 并且默认为 false. 似乎不可能在 $SPARK_HOME/conf/spark-defaults.conf 中使用 全局启用它 spark.sql.caseSensi
..
Spark-submit 找不到本地文件
我已经编写了一个非常简单的 python 脚本来测试我的 Spark 流创意，并计划在我的本地机器上运行它以稍微搞砸一下.这是命令行: spark-submit spark_streaming.py localhost 9999 但是终端给我一个错误: Error execution Jupyter command '': [Errno 2] No such file or director
..
如何在每个执行程序中加载一次文件?
我定义了以下代码以加载预训练的嵌入模型: 导入gensim从 gensim.models.fasttext 导入 FastText 作为 FT_gensim将 numpy 导入为 np类加载器(对象):缓存 = {}emb_dic = {}计数 = 0def __init__(self, 文件名):打印(“|------------------------------------|")打印(“欢
..
如何使用 AWS Glue 运行任意/DDL SQL 语句或存储过程
是否可以从 AWS Glue python 作业执行任意 SQL 命令，例如 ALTER TABLE?我知道我可以用它从表中读取数据，但是有没有办法执行其他数据库特定的命令? 我需要将数据摄取到目标数据库中，然后立即运行一些 ALTER 命令. 解决方案 因此，在进行了广泛的研究并在 AWS 支持下打开了一个案例后，他们告诉我目前无法从 Python shell 或 Glue pys
..
如何在 Apache Spark 中读取包含多个文件的 zip
我有一个包含多个文本文件的压缩文件.我想读取每个文件并构建一个包含每个文件内容的 RDD 列表. val test = sc.textFile("/Volumes/work/data/kaggle/dato/test/5.zip") 将只是整个文件，但如何遍历 zip 的每个内容，然后使用 Spark 将其保存在 RDD 中. 我对 Scala 或 Python 没问题. 在 Py
..
pyspark 解析固定宽度的文本文件
尝试解析固定宽度的文本文件. 我的文本文件如下所示，我需要一个行 ID、日期、一个字符串和一个整数: 00101292017you123400201302017 me5678 我可以使用 sc.textFile(path) 将文本文件读取到 RDD.我可以使用解析后的 RDD 和模式 createDataFrame.这是这两个步骤之间的解析. 解决方案 Spark 的 subst
..
Pyspark 附加执行器环境变量
是否可以在 spark 中为 worker 的 PYTHONPATH 附加值? 我知道可以转到每个工作节点，配置 spark-env.sh 文件并执行此操作，但我想要更灵活的方法 我正在尝试使用 setExecutorEnv 方法，但没有成功 conf = SparkConf().setMaster("spark://192.168.10.11:7077")\.setAppName(
..
从 PySpark 中的数据框中删除重复项
我在本地处理 pyspark 1.4 中的数据帧，并且在使 dropDuplicates 方法工作时遇到问题.它不断返回错误: "AttributeError: 'list' 对象没有属性 'dropDuplicates'" 不太清楚为什么，因为我似乎遵循 最新文档. #将 CSV 文件加载到 RDD 中以开始处理数据rdd1 = sc.textFile("C:\myfilename
..
用户定义的函数要应用于 PySpark 中的 Window?
我正在尝试将用户定义的函数应用于 PySpark 中的 Window.我已经读过 UDAF 可能是要走的路，但我找不到任何具体的东西. 举个例子(取自这里:Xinh's技术博客并针对 PySpark 进行了修改): from pyspark import SparkConf从 pyspark.sql 导入 SparkSession从 pyspark.sql.window 导入窗口从 pys
..