这篇文章主要为大家展示了“Spark 3.0中pandas支持及其与DataFrame相互转换的示例分析”,内容简而易懂,条理清晰,希望能够帮助大家解决疑惑,下面让小编带领大家一起研究并学习一下“Spark 3.0中pandas支持及其与DataFrame相互转换的示例分析”这篇文章吧。
pandas是python用户广泛使用的数据分析库,Spark 3.0已经能较好滴支持pandas接口,从而弥补pandas不能跨机进行大数据处理的不足。pandas还能够与Spark原来的DataFrame相互转换,方便Spark和Python的库相互调用。
Koalas(https://koalas.readthedocs.io/en/latest/)项目使数据科学家在处理大数据时能够更有效率,通过在Spark的上层实现一套pandas DataFrame API。pandas 是python数据处理事实上的标准,而Spark是大数据处理的事实上的标准。通过Koalas,可以:
通过 Spark 立即提升大数据处理生产力,如果熟悉pandas不用学习任何新的知识。
在pandas (tests, smaller datasets) 和 Spark (distributed datasets)只需要一套数据分析代码,方便从研究环境扩展到生产环节。
Koalas要求PySpark,需要首先安装PySpark。
Koalas安装的多种方式包括:
Conda
PyPI
Installation from source
安装PySpark,可以使用:
Installation with the official release channel
Conda
PyPI
Installation from source
建议Python 3.5 及以上版本。
通过 Conda 安装
首先需要安装 Conda ,然后创建一个conda环境。如下:
conda create --name koalas-dev-env
将创建一个只有 Python的最小环境,激活当前环境:
conda activate koalas-dev-env
安装 Koalas:
conda install -c conda-forge koalas
安装Koalas的特定版本:
conda install -c conda-forge koalas=0.19.0
从 PyPI 安装
Koalas 可以使用 pip 从 PyPI 安装:
pip install koalas
从源码安装
查看 Contribution Guide 获得更多指南。
采用官方频道安装:
安装PySpark,从 the official release channel 下载。下载后,解包:
tar xzvf spark-2.4.4-bin-hadoop2.7.tgz
设置 SPARK_HOME
环境变量:
cd spark-2.4.4-bin-hadoop2.7 export SPARK_HOME=`pwd`
确保 PYTHONPATH
可以被 PySpark 和 Py4J找到,在 $SPARK_HOME/python/lib
:
export PYTHONPATH=$(ZIPS=("$SPARK_HOME"/python/lib/*.zip); IFS=:; echo "${ZIPS[*]}"):$PYTHONPATH
从Conda安装:
PySpark 也可以从 Conda 安装:
conda install -c conda-forge pyspark
从PyPI安装:
PySpark 可以从 PyPI 安装:
pip install pyspark
首先,import Koalas 如下:
import pandas as pdimport numpy as npimport databricks.koalas as ksfrom pyspark.sql import SparkSession
创建 Koalas Series,创建一个整数序列值:
s = ks.Series([1, 3, 5, np.nan, 6, 8])
s
0 1.0 1 3.0 2 5.0 3 NaN 4 6.0 5 8.0 dtype: float64
创建 Koalas DataFrame,导入词典对象,转为一个序列:
kdf = ks.DataFrame({'a': [1, 2, 3, 4, 5, 6], 'b': [100, 200, 300, 400, 500, 600], 'c': ["one", "two", "three", "four", "five", "six"]},index=[10, 20, 30, 40, 50, 60])
kdf
a | b | c | |
---|---|---|---|
10 | 1 | 100 | one |
20 | 2 | 200 | two |
30 | 3 | 300 | three |
40 | 4 | 400 | four |
50 | 5 | 500 | five |
60 | 6 | 600 | six |
创建 pandas DataFrame,导入 numpy array,带datetime index 和 labeled columns:
dates = pd.date_range('20130101', periods=6)
dates
DatetimeIndex(['2013-01-01', '2013-01-02', '2013-01-03', '2013-01-04', '2013-01-05', '2013-01-06'], dtype='datetime64[ns]', freq='D')
pdf = pd.DataFrame(np.random.randn(6, 4), index=dates, columns=list('ABCD'))
A | B | C | D | |
---|---|---|---|---|
2013-01-01 | -0.407291 | 0.066551 | -0.073149 | 0.648219 |
2013-01-02 | -0.848735 | 0.437277 | 0.632657 | 0.312861 |
2013-01-03 | -0.415537 | -1.787072 | 0.242221 | 0.125543 |
2013-01-04 | -1.637271 | 1.134810 | 0.282532 | 0.133995 |
2013-01-05 | -1.230477 | -1.925734 | 0.736288 | -0.547677 |
2013-01-06 | 1.092894 | -1.071281 | 0.318752 | -0.477591 |
现在,把pandas DataFrame 转为 Koalas DataFrame:
kdf = ks.from_pandas(pdf)
type(kdf)
databricks.koalas.frame.DataFrame
看起来与 pandas DataFrame几乎一样。
更多例程:https://koalas.readthedocs.io/en/latest/getting_started/10min.html
pandas与dataframe、koalas都可以相互转换。注意pandas与dataframe的转换效率较低,而且pandas原生接口是单机的,建议使用Koalas。
from pyspark.sql import SparkSession# 初始化spark会话spark = SparkSession \ .builder \ .getOrCreate() spark_df = spark.createDataFrame(pandas_df)
import pandas as pdpandas_df = spark_df.toPandas()
由于pandas
的方式是单机版的,即toPandas()
的方式是单机版的,所以参考breeze_lsw改成分布式版本:
import pandas as pddef _map_to_pandas(rdds):return [pd.DataFrame(list(rdds))] def topas(df, n_partitions=None):if n_partitions is not None: df = df.repartition(n_partitions) df_pand = df.rdd.mapPartitions(_map_to_pandas).collect() df_pand = pd.concat(df_pand) df_pand.columns = df.columnsreturn df_pand pandas_df = topas(spark_df)
以上是“Spark 3.0中pandas支持及其与DataFrame相互转换的示例分析”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。