pyspark - 如何在 Pyspark 中检查 Spark 分区中的特定分区数据
问题描述
我在 pyspark 中从我的 hive 表中创建了两个数据框:
data1 = spark.sql("""
SELECT ID, MODEL_NUMBER, MODEL_YEAR ,COUNTRY_CODE
from MODEL_TABLE1 where COUNTRY_CODE in ('IND','CHN','USA','RUS','AUS')
""");
每个国家都有数百万个字母数字格式的唯一 ID。
data2 = spark.sql("""
SELECT ID,MODEL_NUMBER, MODEL_YEAR, COUNTRY_CODE
from MODEL_TABLE2 where COUNTRY_CODE in ('IND','CHN')
""");
我想在 ID 列上使用 pyspark 加入这两个数据框。
我们如何重新分区我们的数据,使其在分区中均匀分布。
我可以使用下面来修复我的数据吗?
newdf1 = data2.repartition(100, "ID")
newdf2 = data2.repartition(100, "ID")
什么是分区的最佳方法,以便更快地加入工作?
解决方案
据我所知,您repartition
提供 ID 列的方法是正确的。考虑以下作为概念证明,spark_partition_id()
用于获取相应的分区 ID:
创建一些虚拟数据
import pandas as pd
import numpy as np
from pyspark.sql.functions import spark_partition_id
def create_dummy_data():
data = np.vstack([np.random.randint(0, 5, size=10),
np.random.random(10)])
df = pd.DataFrame(data.T, columns=["id", "values"])
return spark.createDataFrame(df)
def show_partition_id(df):
"""Helper function to show partition."""
return df.select(*df.columns, spark_partition_id().alias("pid")).show()
df1 = create_dummy_data()
df2 = create_dummy_data()
在重新分区之前显示分区 ID
show_partition_id(df1)
+---+-------------------+---+
| id| values|pid|
+---+-------------------+---+
|1.0| 0.6051170383675885| 0|
|3.0| 0.4613520717857513| 0|
|0.0| 0.797734780966592| 1|
|2.0|0.35594664760134587| 1|
|2.0|0.08223203758144915| 2|
|0.0| 0.3112880092048709| 2|
|4.0| 0.2689639324292137| 3|
|1.0| 0.6466782159542134| 3|
|0.0| 0.8340472796153436| 3|
|4.0| 0.8054752411745659| 3|
+---+-------------------+---+
show_partition_id(df2)
+---+-------------------+---+
| id| values|pid|
+---+-------------------+---+
|4.0| 0.8950517294190533| 0|
|3.0| 0.4084717827425539| 0|
|3.0| 0.798146627431009| 1|
|4.0| 0.8039931522181247| 1|
|3.0| 0.732125135531736| 2|
|0.0| 0.536328329270619| 2|
|1.0|0.25952064363007576| 3|
|2.0| 0.1958334111199559| 3|
|0.0| 0.728098753644471| 3|
|0.0| 0.9825387111807906| 3|
+---+-------------------+---+
重新分区后显示分区ID
show_partition_id(df1.repartition(2, "id"))
+---+-------------------+---+
| id| values|pid|
+---+-------------------+---+
|1.0| 0.6051170383675885| 0|
|3.0| 0.4613520717857513| 0|
|4.0| 0.2689639324292137| 0|
|1.0| 0.6466782159542134| 0|
|4.0| 0.8054752411745659| 0|
|0.0| 0.797734780966592| 1|
|2.0|0.35594664760134587| 1|
|2.0|0.08223203758144915| 1|
|0.0| 0.3112880092048709| 1|
|0.0| 0.8340472796153436| 1|
+---+-------------------+---+
show_partition_id(df2.repartition(2, "id"))
+---+-------------------+---+
| id| values|pid|
+---+-------------------+---+
|4.0| 0.8950517294190533| 0|
|3.0| 0.4084717827425539| 0|
|3.0| 0.798146627431009| 0|
|4.0| 0.8039931522181247| 0|
|3.0| 0.732125135531736| 0|
|1.0|0.25952064363007576| 0|
|0.0| 0.536328329270619| 1|
|2.0| 0.1958334111199559| 1|
|0.0| 0.728098753644471| 1|
|0.0| 0.9825387111807906| 1|
+---+-------------------+---+
重新分区后,id 0 和 2 位于同一个分区上,其余的位于另一个分区上。
推荐阅读
- python - 往返传奇
- java - 使用 append 而不是 LogstashMarker 的 appendRaw 时丢失日志消息
- java - 在 GraphQL 上获取原始字段名称
- optimization - 有没有办法知道在 Gurobi usercuts 中花费了多少时间?
- javascript - 单击导航栏按钮并转到实际的 div?
- python - 是否有一种更简洁的方式来导航这个 Python 嵌套数据结构?
- snowflake-cloud-data-platform - 新帐户命令未将电子邮件识别为字符串
- powershell - 在远程计算机上卸载包
- kotlin - 对 kotlin 的 PMD 支持
- java - Java - 仅删除第一个反斜杠