spark_0">databricks spark基本使用方法
文章目录
spark_dataframepandas_dataframe_3">spark dataframe和pandas dataframe区别
概念
Spark 的 DataFrame 和 pandas 的 DataFrame 在概念上相似，都是用来处理表格数据的，但它们在设计、实现和使用场景上有显著的差异：
Spark DataFrame
1.分布式计算
2.数据存储在集群的多个节点上
3.懒执行（lazy execution）（如调用 .show() 或 .collect() 时）才实际执行。
pandas DataFrame：
1.单机内存中的数据处理
2.操作（如添加列、过滤等）会立即在 DataFrame 上执行并返回结果。
小例子：感受下语法差异！
为了展现差异，下面同样的意思，让两者分别code，感受下语法的差异
spark dataframe
（一般在databricks上面不用建立session，环境已经帮你配置好了）
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Example").getOrCreate()
df = spark.read.csv("data.csv")
df.na.fill(value=0) # 填充数字型缺失值为0
df.na.drop() # 删除任何包含缺失值的行from pyspark.sql.functions import to_date
df.withColumn('new_date', to_date(df['date'], 'yyyy-MM-dd'))from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
def square(x):return x * x
square_udf = udf(square, IntegerType())
df.withColumn('squared', square_udf(df['number']))
pandas dataframe
import pandas as pd
df = pd.read_csv("data.csv")
df.fillna(value=0) # 填充数字型缺失值为0
df.dropna() # 删除任何包含缺失值的行
df['new_date'] = pd.to_datetime(df['date'], format='%Y-%m-%d')
df['squared'] = df['number'].apply(lambda x: x * x)
基本使用
生成序列数据
df1 = spark.range(2, 10, 2)
df2 = spark.range(2, 10, 4)
生成的数据的index名字叫做“id",这里的df1为
2，4，6，8
df2的数据为
2，6
因此将两者join的话
df3 = df1.join(df2, ["id"])
df3的结果为2，6
显示数据
df1.show(10)
不指定的话，默认会展示20条数据
查看rdd的分区数和作用
df3.rdd.getNumPartitions()
作用：
-
并行度评估：RDD的分区数决定了Spark作业的并行度。每个分区通常由一个核心（core）处理，如果分区数太少，可能无法充分利用集群的所有资源；如果分区数过多，则可能因为调度和管理开销而降低性能。
-
性能优化：了解当前的分区数可以帮助你决定是否需要重新分区。通过调整分区数（使用
repartition()或coalesce()方法），来优化作业的性能
对列进行操作
from pyspark.sql.functions import spark_partition_id
df3.withColumn("partition_id", spark_partition_id()).show()
使用spark_partition_id函数可以帮助获得数据所在的分区的id。这里用withColumn之后返回了一个新的对象（rdd不可变，因此每次的操作实际上都会生成新的对象），并且调用show(),把这个对象使用掉了。如果希望是把分区id加上并且存下来，需要写：
from pyspark.sql.functions import spark_partition_id
df3 = df3.withColumn("partition_id", spark_partition_id())
这里，withColumn实际上是DataFrame API的一部分，而不是直接操作RDD。当在DataFrame上使用withColumn方法时，是在定义一个转换操作，这个操作会在DataFrame的执行计划中被添加。虽然DataFrame是建立在RDD之上的，所有DataFrame的操作最终都会转换成对RDD的操作，但从用户的角度看，withColumn是一个更高级别的抽象，专门用于结构化数据的操作。使用DataFrame API可以使代码更易于理解和维护，并且可以利用Spark的优化引擎（如Catalyst优化器和Tungsten执行引擎）来提高性能。
对列的数据进行统计
df2.withColumn("partition_id", spark_partition_id() ).groupBy("partition_id").count().show()
