
!pip install pyspark==3.3.1 py4j==0.10.9.5

Collecting pyspark==3.3.1
  Downloading pyspark-3.3.1.tar.gz (281.4 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 281.4/281.4 MB 2.9 MB/s eta 0:00:00
  Preparing metadata (setup.py) ... done
Collecting py4j==0.10.9.5
  Downloading py4j-0.10.9.5-py2.py3-none-any.whl (199 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 199.7/199.7 kB 15.6 MB/s eta 0:00:00
Building wheels for collected packages: pyspark
  Building wheel for pyspark (setup.py) ... done
  Created wheel for pyspark: filename=pyspark-3.3.1-py2.py3-none-any.whl size=281845494 sha256=29e5202bc43907a8c41b7ffbdfc5a5bf38051efdf687cf46c521d0f4651d78ab
  Stored in directory: /root/.cache/pip/wheels/0f/f0/3d/517368b8ce80486e84f89f214e0a022554e4ee64969f46279b
Successfully built pyspark
Installing collected packages: py4j, pyspark
  Attempting uninstall: py4j
    Found existing installation: py4j 0.10.9.7
    Uninstalling py4j-0.10.9.7:
      Successfully uninstalled py4j-0.10.9.7
Successfully installed py4j-0.10.9.5 pyspark-3.3.1


from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark UDF") \
    .getOrCreate()


columns = ["Seqno","Name"]
data = [("1", "john jones"),
    ("2", "tracey smith"),
    ("3", "amy sanders")]

df = spark.createDataFrame(data=data,schema=columns)

df.show(truncate=False)

+-----+------------+
|Seqno|Name        |
+-----+------------+
|1    |john jones  |
|2    |tracey smith|
|3    |amy sanders |
+-----+------------+


# 첫번째 방법
import pyspark.sql.functions as F
from pyspark.sql.types import *

upperUDF = F.udf(lambda z:z.upper())

# UDF를 적용하여 "Curated Name"이라는 새로운 열 생성
# show(truncate=False): 결과 DataFrame을 출력합니다. truncate=False는
# 출력 결과의 길이를 자르지 않도록 설정합니다.
df.withColumn("Curated Name", upperUDF("Name")) \
  .show(truncate=False)

+-----+------------+------------+
|Seqno|Name        |Curated Name|
+-----+------------+------------+
|1    |john jones  |JOHN JONES  |
|2    |tracey smith|TRACEY SMITH|
|3    |amy sanders |AMY SANDERS |
+-----+------------+------------+


# 두번째 방법
def upper_udf(s):
    return s.upper()


# 리턴은 String
upperUDF = F.udf(upper_udf, StringType())

df.withColumn("Curated Name", upperUDF("Name")) \
   .show(truncate=False)

+-----+------------+------------+
|Seqno|Name        |Curated Name|
+-----+------------+------------+
|1    |john jones  |JOHN JONES  |
|2    |tracey smith|TRACEY SMITH|
|3    |amy sanders |AMY SANDERS |
+-----+------------+------------+


# "Name" 열에 upperUDF라는 사용자 정의 함수(UDF)를 적용
# alias 메서드는 PySpark DataFrame에서 열의 이름을 변경하는 데 사용
df.select("Name", upperUDF("Name").alias("Curated Name")).show()

+------------+------------+
|        Name|Curated Name|
+------------+------------+
|  john jones|  JOHN JONES|
|tracey smith|TRACEY SMITH|
| amy sanders| AMY SANDERS|
+------------+------------+


from pyspark.sql.functions import pandas_udf
import pandas as pd

# 데코레이터는 Pandas UDF를 정의하기 위해 사용됩니다. StringType()은 UDF의 반환 유형을 지정
@pandas_udf(StringType())
# upper_udf_f라는 Pandas UDF를 정의
'''(s: pd.Series) -> pd.Series:: 함수의 매개변수와 반환 유형을 지정합니다.
이 함수는 pd.Series 타입의 매개변수 s를 받아들이고, pd.Series 타입의 값을 반환합니다.
즉, 함수는 pandas Series를 입력으로 받아들이고, pandas Series를 반환'''
def upper_udf_f(s: pd.Series) -> pd.Series:
# s.str.upper()는 pandas Series에 대해 각 문자열 요소를 대문자로 변환하는 연산
    return s.str.upper()


# 위에서 정의한 파이썬 upper 함수를 그대로 사용

upperUDF = spark.udf.register("upper_udf", upper_udf_f)
spark.sql("SELECT upper_udf('aBcD')").show()

+---------------+
|upper_udf(aBcD)|
+---------------+
|           ABCD|
+---------------+


df.select("name", upperUDF("name")).show()

+------------+-----------------+
|        name|upper_udf_f(name)|
+------------+-----------------+
|  john jones|       JOHN JONES|
|tracey smith|     TRACEY SMITH|
| amy sanders|      AMY SANDERS|
+------------+-----------------+


df.createOrReplaceTempView("test")
spark.sql("""
    SELECT name, upper_udf(name) `Curated Name` FROM test
""").show()

+------------+------------+
|        name|Curated Name|
+------------+------------+
|  john jones|  JOHN JONES|
|tracey smith|TRACEY SMITH|
| amy sanders| AMY SANDERS|
+------------+------------+


data = [
    {"a": 1, "b": 2},
    {"a": 5, "b": 5}
]

df = spark.createDataFrame(data)
df.withColumn("c", F.udf(lambda x, y: x + y)("a", "b")).show()

+---+---+---+
|  a|  b|  c|
+---+---+---+
|  1|  2|  3|
|  5|  5| 10|
+---+---+---+


def plus(x, y):
    return x + y

# UDF 등록(register)
plusUDF = spark.udf.register("plus", plus)
spark.sql("SELECT plus(1, 2) sum").show()

+---+
|sum|
+---+
|  3|
+---+


df.withColumn("p", plusUDF("a", "b")).show()

+---+---+---+
|  a|  b|  p|
+---+---+---+
|  1|  2|  3|
|  5|  5| 10|
+---+---+---+


df.createOrReplaceTempView("test")
spark.sql("SELECT a, b, plus(a, b) p FROM test").show()

+---+---+---+
|  a|  b|  p|
+---+---+---+
|  1|  2|  3|
|  5|  5| 10|
+---+---+---+


from pyspark.sql.functions import pandas_udf
import pandas as pd

# Define the UDF
@pandas_udf(FloatType())
def average_udf_f(v: pd.Series) -> float:
    return v.mean()

averageUDF = spark.udf.register('average_udf', average_udf_f)
spark.sql('SELECT average_udf(a) FROM test').show()

+--------------+
|average_udf(a)|
+--------------+
|           3.0|
+--------------+


'''df: PySpark DataFrame입니다. agg: 집계(aggregation)를 수행하는 메서드입니다.
여기서는 averageUDF("b").alias("count")를 통해 "b" 열에 대한 평균을 계산하고,
그 결과를 "count"라는 열로 별칭 지정합니다.'''
df.agg(averageUDF("b").alias("count")).show()

+-----+
|count|
+-----+
|  3.5|
+-----+


arrayData = [
        ('James',['Java','Scala'],{'hair':'black','eye':'brown'}),
        ('Michael',['Spark','Java',None],{'hair':'brown','eye':None}),
        ('Robert',['CSharp',''],{'hair':'red','eye':''}),
        ('Washington',None,None),
        ('Jefferson',['1','2'],{})]

df = spark.createDataFrame(data=arrayData, schema = ['name','knownLanguages','properties'])
df.show()

+----------+-------------------+--------------------+
|      name|     knownLanguages|          properties|
+----------+-------------------+--------------------+
|     James|      [Java, Scala]|{eye -> brown, ha...|
|   Michael|[Spark, Java, null]|{eye -> null, hai...|
|    Robert|         [CSharp, ]|{eye -> , hair ->...|
|Washington|               null|                null|
| Jefferson|             [1, 2]|                  {}|
+----------+-------------------+--------------------+


# knownLanguages 필드를 언어별로 짤라서 새로운 레코드로 생성
from pyspark.sql.functions import explode
df2 = df.select(df.name,explode(df.knownLanguages))
df2.printSchema()
df2.show()

root
 |-- name: string (nullable = true)
 |-- col: string (nullable = true)

+---------+------+
|     name|   col|
+---------+------+
|    James|  Java|
|    James| Scala|
|  Michael| Spark|
|  Michael|  Java|
|  Michael|  null|
|   Robert|CSharp|
|   Robert|      |
|Jefferson|     1|
|Jefferson|     2|
+---------+------+


!wget https://s3-geospatial.s3.us-west-2.amazonaws.com/orders.csv

--2024-01-27 14:44:39--  https://s3-geospatial.s3.us-west-2.amazonaws.com/orders.csv
Resolving s3-geospatial.s3.us-west-2.amazonaws.com (s3-geospatial.s3.us-west-2.amazonaws.com)... 52.218.242.153, 3.5.84.22, 3.5.86.146, ...
Connecting to s3-geospatial.s3.us-west-2.amazonaws.com (s3-geospatial.s3.us-west-2.amazonaws.com)|52.218.242.153|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 89951 (88K) [text/csv]
Saving to: ‘orders.csv’

orders.csv          100%[===================>]  87.84K  --.-KB/s    in 0.08s   

2024-01-27 14:44:40 (1.07 MB/s) - ‘orders.csv’ saved [89951/89951]


!head -5 orders.csv

order_id	items
860196503764	[{"name": "DAILY SPF", "quantity": 1, "id": 1883727790094}]
860292645076	[{"name": "DAILY SPF \u2014 Bundle Set", "quantity": 1, "id": 1883875377166}]
860320956628	[{"name": "DAILY SPF", "quantity": 1, "id": 1883919974414}]
860321513684	[{"name": "DAILY SPF", "quantity": 1, "id": 1883920793614}]


from pyspark.sql.functions import *
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, LongType

order = spark.read.options(delimiter='\t').option("header","true").csv("orders.csv")


order.show()

+------------+--------------------+
|    order_id|               items|
+------------+--------------------+
|860196503764|[{"name": "DAILY ...|
|860292645076|[{"name": "DAILY ...|
|860320956628|[{"name": "DAILY ...|
|860321513684|[{"name": "DAILY ...|
|862930665684|[{"name": "DAILY ...|
|862975819988|[{"name": "DAILY ...|
|862985191636|[{"name": "DAILY ...|
|870939295956|[{"name": "DAILY ...|
|880188063956|[{"name": "DAILY ...|
|933014601940|[{"name": "DAILY ...|
|934065930452|[{"name": "DAILY ...|
|938210722004|[{"name": "DAILY ...|
|944748331220|[{"name": "DAILY ...|
|862843896020|[{"name": "DAILY ...|
|862959763668|[{"name": "DAILY ...|
|870966558932|[{"name": "DAILY ...|
|887936647380|[{"name": "DAILY ...|
|908426477780|[{"name": "DAILY ...|
|921300107476|[{"name": "DAILY ...|
|932229710036|[{"name": "DAILY ...|
+------------+--------------------+
only showing top 20 rows


order.printSchema()

root
 |-- order_id: string (nullable = true)
 |-- items: string (nullable = true)


# 데이터프레임을 이용해서 해보기
struct = ArrayType(
    StructType([
        StructField("name", StringType()),
        StructField("id", StringType()),
        StructField("quantity", LongType())
    ])
)


order.withColumn("item", explode(from_json("items", struct))).show(truncate=False)

+------------+-----------------------------------------------------------------------------+------------------------------------------+
|order_id    |items                                                                        |item                                      |
+------------+-----------------------------------------------------------------------------+------------------------------------------+
|860196503764|[{"name": "DAILY SPF", "quantity": 1, "id": 1883727790094}]                  |{DAILY SPF, 1883727790094, 1}             |
|860292645076|[{"name": "DAILY SPF \u2014 Bundle Set", "quantity": 1, "id": 1883875377166}]|{DAILY SPF — Bundle Set, 1883875377166, 1}|
|860320956628|[{"name": "DAILY SPF", "quantity": 1, "id": 1883919974414}]                  |{DAILY SPF, 1883919974414, 1}             |
|860321513684|[{"name": "DAILY SPF", "quantity": 1, "id": 1883920793614}]                  |{DAILY SPF, 1883920793614, 1}             |
|862930665684|[{"name": "DAILY SPF", "quantity": 1, "id": 1887913672718}]                  |{DAILY SPF, 1887913672718, 1}             |
|862975819988|[{"name": "DAILY SPF", "quantity": 1, "id": 1887985827854}]                  |{DAILY SPF, 1887985827854, 1}             |
|862985191636|[{"name": "DAILY SPF \u2014 Bundle Set", "quantity": 1, "id": 1887999164430}]|{DAILY SPF — Bundle Set, 1887999164430, 1}|
|870939295956|[{"name": "DAILY SPF", "quantity": 1, "id": 1900142264334}]                  |{DAILY SPF, 1900142264334, 1}             |
|880188063956|[{"name": "DAILY SPF", "quantity": 1, "id": 1914170572814}]                  |{DAILY SPF, 1914170572814, 1}             |
|933014601940|[{"name": "DAILY SPF", "quantity": 1, "id": 1995572117518}]                  |{DAILY SPF, 1995572117518, 1}             |
|934065930452|[{"name": "DAILY SPF", "quantity": 1, "id": 1997094813710}]                  |{DAILY SPF, 1997094813710, 1}             |
|938210722004|[{"name": "DAILY SPF", "quantity": 1, "id": 2003241992206}]                  |{DAILY SPF, 2003241992206, 1}             |
|944748331220|[{"name": "DAILY SPF", "quantity": 1, "id": 2013027794958}]                  |{DAILY SPF, 2013027794958, 1}             |
|862843896020|[{"name": "DAILY SPF \u2014 Bundle Set", "quantity": 1, "id": 1887779946510}]|{DAILY SPF — Bundle Set, 1887779946510, 1}|
|862959763668|[{"name": "DAILY SPF", "quantity": 1, "id": 1887960727566}]                  |{DAILY SPF, 1887960727566, 1}             |
|870966558932|[{"name": "DAILY SPF", "quantity": 1, "id": 1900184338446}]                  |{DAILY SPF, 1900184338446, 1}             |
|887936647380|[{"name": "DAILY SPF", "quantity": 1, "id": 1926287851534}]                  |{DAILY SPF, 1926287851534, 1}             |
|908426477780|[{"name": "DAILY SPF", "quantity": 1, "id": 1960246870030}]                  |{DAILY SPF, 1960246870030, 1}             |
|921300107476|[{"name": "DAILY SPF", "quantity": 1, "id": 1979032993806}]                  |{DAILY SPF, 1979032993806, 1}             |
|932229710036|[{"name": "DAILY SPF", "quantity": 1, "id": 1994496901134}]                  |{DAILY SPF, 1994496901134, 1}             |
+------------+-----------------------------------------------------------------------------+------------------------------------------+
only showing top 20 rows


order_items = order.withColumn("item", explode(from_json("items", struct))).drop("items")


order_items.show(5)

+------------+--------------------+
|    order_id|                item|
+------------+--------------------+
|860196503764|{DAILY SPF, 18837...|
|860292645076|{DAILY SPF — Bund...|
|860320956628|{DAILY SPF, 18839...|
|860321513684|{DAILY SPF, 18839...|
|862930665684|{DAILY SPF, 18879...|
+------------+--------------------+
only showing top 5 rows


order_items.printSchema()

root
 |-- order_id: string (nullable = true)
 |-- item: struct (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- id: string (nullable = true)
 |    |-- quantity: long (nullable = true)


order_items.createOrReplaceTempView("order_items")


spark.sql("""
    SELECT order_id, CAST(average_udf(item.quantity) as decimal) avg_count
    FROM order_items
    GROUP BY 1
    ORDER BY 2 DESC""").show(5)

+-------------+---------+
|     order_id|avg_count|
+-------------+---------+
|1816674631892|      500|
|1821860430036|      300|
|2186043064532|      208|
|2143034474708|      200|
|2118824558804|      200|
+-------------+---------+
only showing top 5 rows


spark.sql("""SELECT item.quantity FROM order_items WHERE order_id = '1816674631892'""").show()

+--------+
|quantity|
+--------+
|     500|
+--------+


spark.catalog.listTables()

[Table(name='order_items', database=None, description=None, tableType='TEMPORARY', isTemporary=True),
 Table(name='test', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]


# 많은 함수중 만든 UDF를 확인할수 있음
for f in spark.catalog.listFunctions():
    print(f[0])

!
%
&
*
+
-
/
<
<=
<=>
=
==
>
>=
^
abs
acos
acosh
add_months
aes_decrypt
aes_encrypt
aggregate
and
any
approx_count_distinct
approx_percentile
array
array_agg
array_contains
array_distinct
array_except
array_intersect
array_join
array_max
array_min
array_position
array_remove
array_repeat
array_size
array_sort
array_union
arrays_overlap
arrays_zip
ascii
asin
asinh
assert_true
atan
atan2
atanh
average_udf
avg
base64
bigint
bin
binary
bit_and
bit_count
bit_get
bit_length
bit_or
bit_xor
bool_and
bool_or
boolean
bround
btrim
cardinality
cast
cbrt
ceil
ceiling
char
char_length
character_length
chr
coalesce
collect_list
collect_set
concat
concat_ws
contains
conv
corr
cos
cosh
cot
count
count_if
count_min_sketch
covar_pop
covar_samp
crc32
csc
cume_dist
current_catalog
current_database
current_date
current_timestamp
current_timezone
current_user
date
date_add
date_format
date_from_unix_date
date_part
date_sub
date_trunc
datediff
day
dayofmonth
dayofweek
dayofyear
decimal
decode
degrees
dense_rank
div
double
e
element_at
elt
encode
endswith
every
exists
exp
explode
explode_outer
expm1
extract
factorial
filter
find_in_set
first
first_value
flatten
float
floor
forall
format_number
format_string
from_csv
from_json
from_unixtime
from_utc_timestamp
get_json_object
getbit
greatest
grouping
grouping_id
hash
hex
histogram_numeric
hour
hypot
if
ifnull
ilike
in
initcap
inline
inline_outer
input_file_block_length
input_file_block_start
input_file_name
instr
int
isnan
isnotnull
isnull
java_method
json_array_length
json_object_keys
json_tuple
kurtosis
lag
last
last_day
last_value
lcase
lead
least
left
length
levenshtein
like
ln
locate
log
log10
log1p
log2
lower
lpad
ltrim
make_date
make_dt_interval
make_interval
make_timestamp
make_ym_interval
map
map_concat
map_contains_key
map_entries
map_filter
map_from_arrays
map_from_entries
map_keys
map_values
map_zip_with
max
max_by
md5
mean
min
min_by
minute
mod
monotonically_increasing_id
month
months_between
named_struct
nanvl
negative
next_day
not
now
nth_value
ntile
nullif
nvl
nvl2
octet_length
or
overlay
parse_url
percent_rank
percentile
percentile_approx
pi
plus
pmod
posexplode
posexplode_outer
position
positive
pow
power
printf
quarter
radians
raise_error
rand
randn
random
range
rank
reflect
regexp
regexp_extract
regexp_extract_all
regexp_like
regexp_replace
regr_avgx
regr_avgy
regr_count
regr_r2
repeat
replace
reverse
right
rint
rlike
round
row_number
rpad
rtrim
schema_of_csv
schema_of_json
sec
second
sentences
sequence
session_window
sha
sha1
sha2
shiftleft
shiftright
shiftrightunsigned
shuffle
sign
signum
sin
sinh
size
skewness
slice
smallint
some
sort_array
soundex
space
spark_partition_id
split
split_part
sqrt
stack
startswith
std
stddev
stddev_pop
stddev_samp
str_to_map
string
struct
substr
substring
substring_index
sum
tan
tanh
timestamp
timestamp_micros
timestamp_millis
timestamp_seconds
tinyint
to_binary
to_csv
to_date
to_json
to_number
to_timestamp
to_unix_timestamp
to_utc_timestamp
transform
transform_keys
transform_values
translate
trim
trunc
try_add
try_avg
try_divide
try_element_at
try_multiply
try_subtract
try_sum
try_to_binary
try_to_number
typeof
ucase
unbase64
unhex
unix_date
unix_micros
unix_millis
unix_seconds
unix_timestamp
upper
upper_udf
uuid
var_pop
var_samp
variance
version
weekday
weekofyear
when
width_bucket
window
xpath
xpath_boolean
xpath_double
xpath_float
xpath_int
xpath_long
xpath_number
xpath_short
xpath_string
xxhash64
year
zip_with
|
~

티스토리

UDF 실습