데이터 활용과 투자이야기

# pd (판다스)

import pandas as pd
# import pyspark.pandas as ps

my_list = [[230.1, 37.8, 69.2, 22.1], [44.5, 39.3, 45.1, 10.4],[17.2, 45.9, 69.3, 9.3],[151.5, 41.3, 58.5, 18.5]]
col_name = ['TV', 'Radio', 'Newspaper', 'Sales']

#-- pd --
drop_name = ['Newspaper', 'Sales']
dp = pd.DataFrame(my_list, columns= col_name)
dp.drop(drop_name, axis=1).head(4)

# PS (spark)

import pyspark.pandas as ps
spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
#-- ps --
drop_name = ['Newspaper', 'Sales']
ds.drop(*drop_name).show(4)

5.3.10 추출(Filter)

# pd (판다스)

import pyspark.pandas as ps
from pyspark.sql import SparkSession
# 주의 !! databricks에서 pandas의 csv파일 read하려면 로컬에 파일이 있어야만 가능하기 때문에 교재명령은 불가능
# dp = pd.read_csv('Advertising.csv') 교재예시
# 그래서 편법으로 아래와 같이 사용함

import pandas as pd

df_pandas = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True).toPandas()
df_pandas.head(4)

df_pandas['TV'] = df_pandas['TV'].astype(float)
df_pandas['Radio'] = df_pandas['Radio'].astype(float)
df_pandas['Newspaper'] = df_pandas['Newspaper'].astype(float)
df_pandas['Sales'] = df_pandas['Sales'].astype(float)

# df_pandas.dtypes
df_pandas[df_pandas.Newspaper<20].head(4)
df_pandas[(df_pandas.Newspaper<20)&(df_pandas.TV>100)].head(4)

# PS (spark)

ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
ds[ds.Newspaper<20].show(4)
ds[(ds.Newspaper<20)&(ds.TV>100)].show(4)

5.3.11 새 컬럼 추가하기

#-- df_pandas
df_pandas['tv_norm'] = df_pandas.TV/sum(df_pandas.TV)
df_pandas.head(4)

#-- ps
import pyspark.sql.functions as F
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
ds.withColumn('tv_norm', ds.TV/ds.groupBy().agg(F.sum("TV")).collect()[0][0]).show(4)

#-- df_pandas
df_pandas['cond'] = df_pandas.apply(lambda A: 1 if((A.TV>100)&(A.Radio<40)) else 2 if(A.Sales> 10) else 3,axis=1)
df_pandas.head(4)

#-- ps
ds.withColumn('cond', F.when((ds.TV>100)&(ds.Radio<40),1)\
    .when(ds.Sales>10, 2)\
    .otherwise(3)).show(4)

#-- df_pandas
import numpy as np
df_pandas['log_tv'] = np.log(df_pandas.TV)
df_pandas.head(4)

#-- ps
import pyspark.sql.functions as F
ds.withColumn('log_tv', F.log(ds.TV)).show(4)

#-- df_pandas
df_pandas['tv+10'] = df_pandas.TV.apply(lambda x: x+10)
df_pandas.head(4)

#-- ps
ds.withColumn('tv+10', ds.TV+10).show(4)

# 모든 열의 이름 바꾸기

import pandas as pd
import pyspark.pandas as ps
from pyspark.sql import SparkSession

my_list = [[230.1, 37.8, 69.2, 22.1], [44.5, 39.3, 45.1, 10.4],[17.2, 45.9, 69.3, 9.3],[151.5, 41.3, 58.5, 18.5]]
col_name = ['TV', 'Radio', 'Newspaper', 'Sales']
# df
#-- pd --
dp = pd.DataFrame(my_list, columns= col_name)
dp.columns = ['a', 'b', 'c', 'd']
dp.head(4)

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
#-- ps --
ds.toDF('a', 'b', 'c', 'd').show(4)

# 하나 혹은  그 이상의 열 이름 바꾸기

mapping = {'Newspaper':'C', 'Sales':'D'}
#-- pd --
my_list = [[230.1, 37.8, 69.2, 22.1], [44.5, 39.3, 45.1, 10.4],[17.2, 45.9, 69.3, 9.3],[151.5, 41.3, 58.5, 18.5]]
col_name = ['TV', 'Radio', 'Newspaper', 'Sales']
#-- pd --
dp = pd.DataFrame(my_list, columns= col_name)
dp.rename(columns=mapping).head(4)

#-- ps --
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
mapping = {'Newspaper':'C', 'Sales':'D'}
new_names = [mapping.get(col, col) for col in ds.columns]
ds.toDF(*new_names).show(4)

Note : PySpark에서 한 컬럼의 이름을 바꾸기 위해 withColumnRenamed를 사용 할 수도 있습니다.

#-- ps --
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
ds.withColumnRenamed('Newspaper','Paper').show(4)

# import pandas as pd  # bricks는 import가 없어도 됨
# import pyspark.pandas as ps # bricks는 import가 없어도 됨

my_list = [['male', 1, None], ['female', 2, 3], ['male', 3, 4]]
#-- pd --
dp = pd.DataFrame(my_list, columns=['A', 'B', 'C'])
dp.head()

#-- ps --
ds = spark.createDataFrame(my_list, ['A', 'B', 'C'])
ds.show()

#-- pd --
dp.fillna(-99)
#-- ps --
ds.fillna(-99).show()

#-- pd --
# 주의 : 특정 컬럼을 선택해야 합니다.
dp.A.replace(['male','female'],['1', '0'], inplace=True)
dp
#-- ps --
# 주의 : na를 대체하면서 동시에 특정 열의 값을 대체하지 못합니다.
# 이 경우 특정 열의 대체 값만 진행됩니다.
ds.na.replace(['male', 'female'],['1', '0']).show()

# 아래 내용의 출처를 밝힙니다. 파이썬으로 배우는 아파치 스파크 ( Wenqiang Feng, 홍은희 )

RDD

: 복원 분산 데이터 집합(Resilient Distributed Dataset)를 의미합니다.

스파크의 RDD는 불변 객체 집합의 분산 컬렉션(collection)입니다.
각 RDD는 여러 개의 파티션(더 작은 집합과 비슷한 패턴)으로 나뉘며 클러스트의 다른 노드에서 계산될 수 있습니다.

# RDD 생성_00 기본형

from pyspark.sql import SparkSession

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

df = spark.sparkContext.parallelize([(1,2,3,'a b c'),
            (4,5,6,'d e f'),
            (7,8,9,'g h i')]).toDF(['col1', 'col2', 'col3', 'col4'])
df.show()

# RDD 생성_01 createDataFrame를 이용한 생성, 컬럼명 명시

from pyspark.sql import SparkSession

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

Employee = spark.createDataFrame([
    ('1', 'Joe', '70000', '1'),
    ('2', 'Henry', '80000', '2'),
    ('3', 'Sam', '60000', '2'),
    ('4', 'Max', '90000', '1')],
    ['Id', 'Name', 'Sallary', 'DepartmentId']
    )
Employee.show()

# RDD 생성_02 

from pyspark.sql import SparkSession

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

df = spark.read.format('com.databricks.spark.csv'). \
    options(header='true',inferschema='true'). \
    load("/home/feng/Spark/Code/data/Adventising.csv", header=True)

df.show(5)
df.printSchema()

# RDD spark 버전 확인

from pyspark.sql import SparkSession

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()
    
print('Apache Spark 버전 확인: '+spark.version)

결과>>  Apache Spark 버전 확인: 3.5.0

# RDD 생성_03 CSV 파일 load

from pyspark.sql import SparkSession

spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

df = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
df.show(5)
df.printSchema()

팁 ) databricks에서 제공하는 클라우드 spark환경에서 테이블 생성하기 위해 파일 업로드하는 위치를 지정가능함

파일올리면서 테이블을 생성해야 하고 그 위치는 /FileStore/tables/파일명

# RDD 생성_04 외부의 mysql 접속해서 테이블 조회하기

driver = "org.mariadb.jdbc.Driver"
database_host = "xxx.xxx.xxx.xxx"
database_port = "3306" # update if you use a non-default port
database_name = "xxxxxx"
table = "Advertising"
user = "xxxxx"
password = "xxxxx"
url = f"jdbc:mysql://{database_host}:{database_port}/{database_name}"
print(url)

df = (spark.read
.format("jdbc")
.option("driver", driver)
.option("url", url)
.option("query", "select * from Advertising")
.option("user", user)
.option("password", password)
.option("allowPublicKeyRetrieval", True)
.option("useSSL", False)
.load()
)
df.show(5)

팁 )  아래 두개를 .option에 넣어줘야 함

• allowPublicKeyRetrieval=true
• useSSL=false

# RDD 생성_05 hdfs 조회하기

from pyspark.conf import SparkConf
from pyspark.context import SparkContext
from pyspark.sql import HiveContext

sc = SparkContext.getOrCreate();
hc = HiveContext(sc)
tf1 = sc.textFile("hdfs://FileStore/tables/Advertising.csv")
    
print(tf1.first())
# 아직 코딩이 완성된 것이 아님