<jupyter_start><jupyter_text>spark_env 가상환경에서의 빅데이터  역할분담으로 인한 시간 절약<jupyter_code>import numpy as np
import time
sc= SparkContext.getOrCreate()   #스파크 이용 하겠다. 
sc
MAX = 100000
rdd = sc.parallelize([i for i in range(3, MAX+1)],10)
# rdd는 스파크의 인메모리 환경.   # 뒤에 숫자10은 나누어줄 갯수
def get_prime(num):
    isPrime = True
    for index in range(2, num):
        if num% index ==0:
            isPrime == False
            break
            
    if isPrime == True:
        return num
start = time.time()
prime_list = rdd.map(get_prime).collect()  #결과 값을 다 모으는 함수 
end = time.time()
print("소요시간=", end-start)
prime_list
from pyspark.mllib.tree import DecisionTree, DecisionTreeModel
df = spark.read.csv( "playtennis.csv", header=True)
type(df)
df.show()
from pyspark.sql import functions as F
# df.withColumn : 컬럼 데이터를 수정해서 리턴
# df.withColun("Outlook", "F.when(F.col("Outlook")=="Sunny",0))
# Outlook 컬럼의 값이 Sunny이면 0으로 수정해서 리턴
# otherwise(F.col("Outlook")) : 조건이 거짓이면 Outlook 컬럼의 기존 값 리턴
df = df.withColumn("Outlook", F.when(F.col("Outlook")=="Sunny", 0).otherwise(F.col("Outlook")))
df = df.withColumn("Outlook", F.when(F.col("Outlook")=="Overcast",1).otherwise(F.col("Outlook")))
df = df.withColumn("Outlook", F.when(F.col("Outlook")=="Rain",2).otherwise(F.col("Outlook")))

df = df.withColumn("Temperature", F.when(F.col("Temperature")=="Hot",0).otherwise(F.col("Temperature")))
df = df.withColumn("Temperature", F.when(F.col("Temperature")=="Mild",1).otherwise(F.col("Temperature")))
df = df.withColumn("Temperature", F.when(F.col("Temperature")=="Cool",2).otherwise(F.col("Temperature")))

df = df.withColumn("Humidity", F.when(F.col("Humidity")=="High",0).otherwise(F.col("Humidity")))
df = df.withColumn("Humidity", F.when(F.col("Humidity")=="Normal",1).otherwise(F.col("Humidity")))

df = df.withColumn("Wind", F.when(F.col("Wind")=="Weak",0).otherwise(F.col("Wind")))
df = df.withColumn("Wind", F.when(F.col("Wind")=="Strong",1).otherwise(F.col("Wind")))

df = df.withColumn("PlayTennis", F.when(F.col("PlayTennis")=="No",0).otherwise(F.col("PlayTennis")))
df = df.withColumn("PlayTennis", F.when(F.col("PlayTennis")=="Yes",1).otherwise(F.col("PlayTennis")))<jupyter_output><empty_output><jupyter_text>위에 처럼 각각 지정해주는 방법과, spark 에서보다, Pandas에서 일괄 처리해주고, 다시 spark 변환해주는 것이 시간효율성을 높일수 있다.<jupyter_code>df.show()
df.schema          # 컬럼 명, 타입 을 조회하는 함수<jupyter_output><empty_output><jupyter_text>밑의 과정은, 값 타입을 변경해주는 것이며, 판다스에서도 동일한 과정을 거처야한다.(시간은 같음)<jupyter_code>df= df.withColumn("Outlook", df["Outlook"].cast('int'))

# withColumn이라는 것은 각 블록을 돌아다니며 새것으로 덮어 씌우는 것으로 이해하면 된다.<jupyter_output><empty_output><jupyter_text>![image.png](attachment:image.png)<jupyter_code>df = df.withColumn("Temperature", df["Temperature"].cast('int'))
df = df.withColumn("Humidity", df["Humidity"].cast('int'))
df = df.withColumn("Wind", df["Wind"].cast('int'))
df = df.withColumn("PlayTennis", df["PlayTennis"].cast('int'))
df.show()<jupyter_output>+-------+-----------+--------+----+----------+
|Outlook|Temperature|Humidity|Wind|PlayTennis|
+-------+-----------+--------+----+----------+
|      0|          0|       0|   0|         0|
|      0|          0|       0|   1|         0|
|      1|          0|       0|   0|         1|
|      2|          1|       0|   0|         1|
|      2|          2|       1|   0|         1|
|      2|          2|       1|   1|         0|
|      1|          2|       1|   1|         1|
|      0|          1|       0|   0|         0|
|      0|          2|       1|   0|         1|
|      2|          1|       1|   0|         1|
|      0|          1|       1|   1|         1|
|      1|          1|       0|   1|         1|
|      1|          0|       1|   0|         1|
|      2|          1|       0|   1|         0|
+-------+-----------+--------+----+----------+<jupyter_text>위에까지의 과정은, 문자를 숫자로, 숫자(str) 타입을 int로 변경해주는 과정이었다  이제부터는, spark를 이용한 Decision Tree를 만들어보자<jupyter_code>from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=["Outlook", "Temperature", "Humidity", "Wind"], outputCol="features")
assembler_df = assembler.transform(df)  # assembler의 features 은 위의 inputCols의 컬럼값을 합쳐서 생성됨,
assembler_df.show()<jupyter_output>+-------+-----------+--------+----+----------+-----------------+
|Outlook|Temperature|Humidity|Wind|PlayTennis|         features|
+-------+-----------+--------+----+----------+-----------------+
|      0|          0|       0|   0|         0|        (4,[],[])|
|      0|          0|       0|   1|         0|    (4,[3],[1.0])|
|      1|          0|       0|   0|         1|    (4,[0],[1.0])|
|      2|          1|       0|   0|         1|[2.0,1.0,0.0,0.0]|
|      2|          2|       1|   0|         1|[2.0,2.0,1.0,0.0]|
|      2|          2|       1|   1|         0|[2.0,2.0,1.0,1.0]|
|      1|          2|       1|   1|         1|[1.0,2.0,1.0,1.0]|
|      0|          1|       0|   0|         0|    (4,[1],[1.0])|
|      0|          2|       1|   0|         1|[0.0,2.0,1.0,0.0]|
|      2|          1|       1|   0|         1|[2.0,1.0,1.0,0.0]|
|      0|          1|       1|   1|         1|[0.0,1.0,1.0,1.0]|
|      1|          1|       0|   1|         1|[1.0,1.0,0.0,1.0]|
|      1|          0|    [...]<jupyter_text>여기에서 부터, 데이터를 분리하는 작업<jupyter_code>(trainingData, testData) = assembler_df.randomSplit([0.7, 0.3])<jupyter_output><empty_output><jupyter_text>![image.png](attachment:image.png)   Classifier<jupyter_code>from pyspark.ml.classification import DecisionTreeClassifier
dt= DecisionTreeClassifier(featuresCol= "features", labelCol = "PlayTennis")
dtModel= dt.fit(trainingData)
dtModel<jupyter_output><empty_output><jupyter_text>![image.png](attachment:image.png)<jupyter_code>prediction= dtModel.transform(testData)
prediction.show()
right_predict =prediction.filter(prediction.PlayTennis == prediction.prediction)
right_predict.show()
a=right_predict.count()
b=prediction.count()
print(a, b)
# 정확도 계산 = right_predict.count() / prediction.count() 
# 위에서 각각 a,b로 변수 지정해주었으니 계산은 a/b
print("정확도=", a/b)<jupyter_output>정확도= 0.8333333333333334<jupyter_text>시각화 하기<jupyter_code>from dtreeviz import trees
from dtreeviz.models.spark_decision_tree import ShadowSparkTree
# 독립변수를 조회 하기 
trainingData["Outlook","Temperature","Humidity","Wind"].show()<jupyter_output>+-------+-----------+--------+----+
|Outlook|Temperature|Humidity|Wind|
+-------+-----------+--------+----+
|      0|          0|       0|   0|
|      0|          1|       0|   0|
|      0|          2|       1|   0|
|      1|          0|       0|   0|
|      1|          0|       1|   0|
|      1|          1|       0|   1|
|      2|          2|       1|   0|
|      2|          2|       1|   1|
+-------+-----------+--------+----+<jupyter_text>위에 다중선택은 대괄호가 1개, 밑에 1개 선택은 대괄호가 2개 인 이유는 무엇?위의 같은 경우 다중으로, 프레임이 만들어지는 경우이지만, 1개인 경우 내용만 출력되기 때문에 괄호를 쓰여주여한다. [[]]2개인것은 데이터 프레임 .<jupyter_code>trainingData[["PlayTennis"]].show()<jupyter_output>+----------+
|PlayTennis|
+----------+
|         0|
|         0|
|         1|
|         1|
|         1|
|         1|
|         1|
|         0|
+----------+<jupyter_text>위에서 만든 dtModel 의 내요을 조회하는 부분으로, 어떻게 분류가 되었는지 알 수 있는 부분이다.<jupyter_code>print(dtModel.toDebugString)   # depth는 3인걸 위에서 먼저 확인 할 수 있었다.
trainingData.show()
dataset=trainingData.toPandas()  # 여러개로 분산되어 있는 것을 하나로 모으기 위해
                                #pandas로 프레임 만들어주기. 
dataset
dataset[["Outlook","Temperature","Humidity","Wind", "PlayTennis"]]<jupyter_output><empty_output><jupyter_text>Spark Decision Tree 시각화 할 객체 = shadowSparkTree<jupyter_code>spark_dtree = ShadowSparkTree (
                                dtModel,
                                dataset[["Outlook","Temperature","Humidity","Wind"]],
                                dataset["PlayTennis"],
                                feature_names=["Outlook","Temperature","Humidity","Wind"],
                                target_name="PlayTennis",
                                class_names=["No","Yes"]
)
spark_dtree
import os
os.environ["PATH"]+=os.pathsep + "c:/Program Files/Graphviz/bin"
os.environ["PATH"]<jupyter_output><empty_output><jupyter_text>![image.png](attachment:image.png)  Decision 내용 시각화<jupyter_code>trees.dtreeviz(spark_dtree, scale=2.0)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.20888238164958517, 'fraction_numerical': 0.03733528550512445, 'mean_word_length': 2.1648648648648647, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '426248', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 3622, 'n_tokens_neox': 3829, 'n_words': 566, 'path': 'Machine_Learning/Spark_Numpy.ipynb', 'repo_name': 'jeewoolee-jasmine/TIL', 'size': '8196'}