PySpark 如何将 CSV 读入 Dataframe，并对其进行操作
我对 pyspark 很陌生，我正在尝试使用它来处理一个保存为 csv 文件的大型数据集.我想将 CSV 文件读入 spark 数据框，删除一些列，然后添加新列.我该怎么做? 我无法将此数据放入数据框中.这是我目前所拥有的精简版: def make_dataframe(data_portion, schema, sql):字段 = data_portion.split(",")返回 sql
..
pyspark相关内容
使用 Apache Spark 将键值对缩减为键列表对
我正在编写一个 Spark 应用程序，并希望将一组键值对 (K, V1), (K, V2), ..., (K, Vn) 组合成一个键-多值对(K, [V1, V2, ..., Vn]).我觉得我应该能够使用具有某种风味的 reduceByKey 函数来做到这一点: My_KMV = My_KV.reduce(lambda a, b: a.append([b])) 发生这种情况时我得到的错误是:
..
从数据框中获取价值
在 Scala 中，我可以使用 get(#) 或 getAs[Type](#) 从数据帧中获取值.在 pyspark 中应该怎么做? 我有一个两列 DataFrame:item(string) 和 salesNum(integers).我做了一个 groupby 和 mean 来获得这些数字的平均值，如下所示: saleDF.groupBy("salesNum").mean()).co
..
如何在 Pyspark 中随时间序列数据使用滑动窗口转换数据
我正在尝试根据时间序列数据的滑动窗口提取特征.在 Scala 中，似乎有一个基于 this post 和 文档 import org.apache.spark.mllib.rdd.RDDFunctions._sc.parallelize(1 到 100, 10).滑动(3).map(curSlice => (curSlice.sum/curSlice.size)).搜集() 我的问题是 PyS
..
如何在 anaconda 中导入 pyspark
我正在尝试将 pyspark 与 anaconda 一起导入和使用. 安装 spark 后，我尝试设置 $SPARK_HOME 变量: $ pip install pyspark (当然)这行不通，因为我发现我需要通过 tel python 在 $SPARK_HOME/python/ 下查找 pyspark.问题是要做到这一点，我需要设置 $PYTHONPATH 而 anaconda 不
..
配置 Spark 以使用 Jupyter Notebook 和 Anaconda
我花了几天时间尝试让 Spark 与我的 Jupyter Notebook 和 Anaconda 一起工作.这是我的 .bash_profile 的样子: PATH="/my/path/to/anaconda3/bin:$PATH"导出 JAVA_HOME=“/my/path/to/jdk"导出 PYTHON_PATH=“/my/path/to/anaconda3/bin/python"导出 P
..
PySpark - 在数据框中求和一列并将结果返回为 int
我有一个带有一列数字的 pyspark 数据框.我需要对该列求和，然后将结果返回为 python 变量中的 int. df = spark.createDataFrame([("A", 20), ("B", 30), ("D", 80)],["Letter", "Number"]) 我执行以下操作来对列求和. df.groupBy().sum() 但我得到了一个数据框. +-------
..
pyspark中的probnorm函数等效
PROBNORM:解释 SAS 中的 PROBNORM 函数返回标准正态分布的观测值小于或等于 x 的概率. pyspark中有没有等价的功能? 解决方案 恐怕PySpark中没有这样的实现方法. 但是，您可以利用 Pandas UDF 使用基本的 Python 包定义您自己的自定义函数！这里我们将使用 scipy.stats.norm 模块从标准正态分布中获取累积概率.
..
java.io.IOException:无法运行程序“python"在 Pycharm (Windows) 中使用 Spark
我正在尝试在 Pycharm 中使用 Spark 编写一个非常简单的代码，而我的操作系统是 Windows 8.我一直在处理几个问题，除了一个之外，这些问题都设法解决了.当我使用 pyspark.cmd 运行代码时，一切正常，但我在 pycharm 中使用相同的代码没有运气.我使用以下代码修复了 SPARK_HOME 变量存在问题: 导入系统导入操作系统os.environ['SPARK_HOM
..
AttributeError: 'NoneType' 对象没有属性 'setCallSite'
在 PySpark 中，我想计算两个数据帧向量之间的相关性，使用以下代码(我在导入 pyspark 或 createDataFrame 时没有任何问题): from pyspark.ml.linalg import Vectors从 pyspark.ml.stat 导入相关性导入pysparkspark = pyspark.sql.SparkSession.builder.master("loc
..
我如何使用 spark-env.sh.template 在 Python 3 中设置 Pyspark
因为我的 ipython3 笔记本中有这个问题，我想我必须以某种方式更改“spark-env.sh.template". 例外:worker 中的 Python 2.7 版本与驱动程序 3.4 中的版本不同，PySpark 无法在不同的次要版本下运行 解决方案 Spark 尚不能与 Python 3 配合使用.如果您希望使用 Python API，您还需要一个 Python 解释器(
..
将 PySpark 与 Jupyter Notebook 集成
我正在关注此站点以安装 Jupyter Notebook、PySpark 和整合两者. 当我需要创建“Jupyter 配置文件"时，我读到“Jupyter 配置文件"不再存在.所以我继续执行以下几行. $ mkdir -p ~/.ipython/kernels/pyspark$ touch ~/.ipython/kernels/pyspark/kernel.json 我打开了 kerne
..
jupyter 抛出错误:socket.gaierror: [Errno -2] 名称或服务未知
我在 AWS EMR 上运行了一个 pyspark 集群，并使用 Jupyter 来运行 pyspark 驱动程序.最近，脚本失败了.Jupiter 无法启动服务器.我用 conda install jupyter 安装，并从 sudo initctl start jupyter 开始.集群运行良好.这是服务器的配置值. # jupyter 配置mkdir -p ~/.jupyter触摸 ls
..
Spark:PySpark + Cassandra 查询性能
我在本地机器(8 核，16GB 内存)上设置了 Spark 2.0 和 Cassandra 3.0 用于测试目的，并编辑了 spark-defaults.conf 如下: spark.python.worker.memory 1gspark.executor.cores 4spark.executor.instances 4spark.sql.shuffle.partitions 4 接下来我
..
将 Spark Structure 流数据写入 Cassandra
我想使用 Pyspark API 将结构流数据写入 Cassandra. 我的数据流如下: Nifi -> Kafka -> Spark Structure Streaming -> Cassandra 我尝试过以下方式: query = df.writeStream\.format("org.apache.spark.sql.cassandra")\.option("keys
..
如何从 PySpark MultilayerPerceptronClassifier 获得分类概率?
我在 python 中使用 Spark 2.0.1，我的数据集在 DataFrame 中，所以我使用 ML(不是 MLLib)库进行机器学习.我有一个多层感知器分类器，但只有两个标签. 我的问题是，是否有可能不仅获得标签，而且(或仅)获得该标签的概率?不仅仅是每个输入的 0 或 1，而是 0.95 表示 0 和 0.05 表示 1.如果 MLP 无法做到这一点，但其他分类器可以做到，我可以更
..
如何在 Apache Spark 中将时区设置为 UTC?
在 Spark 的 WebUI(端口 8080)和环境选项卡上，有以下设置: user.timezone 祖鲁语 你知道我如何/在哪里可以将其覆盖为 UTC? 环境细节: Spark 2.1.1 jre-1.8.0-openjdk.x86_64 没有 jdk EC2 亚马逊 Linux 编辑(有人回答了下面然后删除):https://www.timeanddate.co
..
通过 Hadoop 输入格式示例用于 pyspark 的 BigQuery 连接器
我有一个存储在 BigQuery 表中的大型数据集，我想将其加载到 pypark RDD 中以进行 ETL 数据处理. 我意识到 BigQuery 支持 Hadoop 输入/输出格式 https://cloud.google.com/hadoop/writing-with-bigquery-connector 和 pyspark 应该能够使用这个接口来通过使用“newAPIHad
..
具有不同列的 Pysaprk 多组分组
我有如下数据 年份名称百分比性别1880 年约翰 0.081541 男孩1881 年威廉 0.080511 男孩1881 年约翰 0.050057 男孩 我需要使用不同的列进行分组和计数 df_year = df.groupby('year').count()df_name = df.groupby('name').count()df_sex = df.groupby('sex').coun
..
查询 Hive 表时，数据帧 NumberFormatException 上的 Spark 2.2 Thrift 服务器错误
我有运行 Spark2 (v2.2) 的 Hortonworks HDP 2.6.3.我的测试用例很简单: 用一些随机值创建一个 Hive 表.Hive 在 10000 端口 在 10016 开启 Spark Thrift 服务器 运行pyspark并通过10016查询Hive表 但是，由于 NumberFormatException，我无法从 Spark 获取数据.
..