? ? PySpark 是 Apache Spark 的 Python API，它提供了使用 Python 編程語言處理大數(shù)據(jù)的能力。Spark 是一個基于內(nèi)存的分布式計算框架，能夠處理大規(guī)模數(shù)據(jù)集，包括結(jié)構(gòu)化數(shù)據(jù)、半結(jié)構(gòu)化數(shù)據(jù)和非結(jié)構(gòu)化數(shù)據(jù)等各種數(shù)據(jù)類型。下面是 PySpark 的一些主要用途：
? ? 數(shù)據(jù)清洗和處理：可以使用 PySpark 處理大規(guī)模的數(shù)據(jù)，對數(shù)據(jù)進(jìn)行清洗、篩選、轉(zhuǎn)換、合并等操作，以便進(jìn)一步的數(shù)據(jù)分析或建模。
? ? 數(shù)據(jù)分析和挖掘：使用 PySpark 可以對大規(guī)模數(shù)據(jù)進(jìn)行探索性數(shù)據(jù)分析 (EDA)、統(tǒng)計分析、機(jī)器學(xué)習(xí)和深度學(xué)習(xí)等操作，以獲取數(shù)據(jù)的更多洞察。
? ? 流式數(shù)據(jù)處理：PySpark 支持 Spark Streaming，可以實時處理流式數(shù)據(jù)，如實時日志分析、流式事件處理等。
? ? 圖計算：使用 PySpark 可以進(jìn)行圖計算，以解決圖結(jié)構(gòu)相關(guān)的問題，如社交網(wǎng)絡(luò)分析、路徑優(yōu)化等。
? ? 批處理任務(wù)：PySpark 可以用于大規(guī)模的批處理任務(wù)，如數(shù)據(jù)導(dǎo)入、數(shù)據(jù)導(dǎo)出、ETL、數(shù)據(jù)清洗、數(shù)據(jù)預(yù)處理等。
以下是一些常用的PySpark查詢語句：
1、選擇所有列：
df.select("*")
2、選擇指定列：
df.select("column1", "column2", "column3")
3、過濾行：
df.filter(df["column1"] > 10)
4、按列排序：
df.sort("column1")
5、按列分組并計算聚合函數(shù)：
df.groupBy("column1").agg({"column2": "sum"})
6、使用 SQL 語句查詢：
df.createOrReplaceTempView("table_name")
spark.sql("SELECT * FROM table_name WHERE column1 > 10")
7、連接兩個數(shù)據(jù)框：
df1.join(df2, df1["column1"] == df2["column2"], "inner")
8、描述數(shù)據(jù)框中的統(tǒng)計信息：
df.describe()
9、添加新的列：
df.withColumn("new_column", df["column1"] + df["column2"])
? ? 這些語句只是 PySpark 中可用的查詢語句的一小部分。
? ? 要使用 PySpark 查詢數(shù)據(jù)庫中的表數(shù)據(jù)，需要先配置 SparkSession 對象以連接到數(shù)據(jù)庫，然后使用 DataFrame API 進(jìn)行查詢。以下是一個簡單的例子：
from pyspark.sql import SparkSession
# 創(chuàng)建 SparkSession 對象
spark = SparkSession.builder \
? ? .appName("Query Table Data") \
? ? .config("spark.driver.extraClassPath", "/path/to/jdbc-driver.jar") \
? ? .getOrCreate()
# 連接數(shù)據(jù)庫
jdbc_url = "jdbc:postgresql://hostname:port/database_name"
user = "username"
password = "password"
table_name = "table"
# 讀取表數(shù)據(jù)
df = spark.read \
? ? .format("jdbc") \
? ? .option("url", jdbc_url) \
? ? .option("user", user) \
? ? .option("password", password) \
? ? .option("dbtable", table_name) \
? ? .load()
# 顯示前 10 行數(shù)據(jù)
df.show(10)
? ? 在上面的代碼中，我們首先創(chuàng)建了一個 SparkSession 對象，然后使用它連接到 PostgreSQL 數(shù)據(jù)庫。接下來，我們使用 DataFrame API 讀取了表 table1 的數(shù)據(jù)，并使用 show() 方法顯示了前 10 行數(shù)據(jù)。根據(jù)你的實際需求，你可以使用不同的查詢方式來獲取所需的數(shù)據(jù)。例如，你可以使用 select() 方法選擇特定的列，或者使用 filter() 方法過濾行。
