Spark Introduction
Published: 2020-12-17 09:33:43

<h1><strong>Quick View</strong></h1>

Spark의 기본에 대해서 정리해 보자.
<h2>RDD</h2>
<ul>
 	<li>RDD는 분산 자료구조이다 ≈ 일단 DB 테이블로 생각하자</li>
 	<li>Mutable 특성 ≈ 테이블인데 append만 되는 read-only table이라고 생각하자</li>
 	<li>Lazy execution 특성
<ul>
 	<li>RDD "Actions"은 operator이며 결과로 새로운 RDD들을 만들어 낸다.</li>
 	<li>RDD actions 체인을 거쳐서 최종 RDD를 설계한 다음,</li>
 	<li>"Transformations"이 실제로 RDD의 actions들을 실행시킨다.</li>
</ul>
</li>
 	<li>DAG (Directed Acyclic Graph)
<ul>
 	<li>RDD의 actions / transformation은 결국 DAG를 설계하는 것이다.</li>
</ul>
</li>
 	<li>작업 단위
<ul>
 	<li>크기 순: Job  &gt; Stage &gt; Task</li>
 	<li>Stage: shuffle이 발생하기 전까지 하나의 executor에 의해서 실행되는 구간</li>
 	<li>Task: map(), filter()와 같은 단위 연산</li>
</ul>
</li>
</ul>
<h2>SparkSQL</h2>
<ul>
 	<li>RDD에 DB table처럼 schema를 부여하여 SQL 질의를 가능하게 할 수 있다. 이러한 RDD를 "DataFrames"라고 부른다.</li>
</ul>
<h2>Spark Streaming</h2>
<ul>
 	<li>Data Source (Kafka/S3/Kinesis/…) → Spark Streaming → Data Destination (Cassandra/HDFS/…)</li>
 	<li>DStream (Discretized Stream)
<ul>
 	<li>Stream block을 5초 동안 모아서 하나의 RDD로 만들어 낸 것이다. 결국 Spark Stream도 RDD 단위로 처리하는 셈</li>
</ul>
</li>
</ul>

<h1><strong>Spark Installation</strong></h1>

<!-- wp:heading -->
<h2>Installing PySpark on Windows</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>설치 가이드를 따라 진행하다가 막히는 부분들을 여기저기 찾으며 정리한 것이다. Anaconda는 미리 설치되어 있다고 가정한다. Spark는 HDFS 파일시스템 위에서 동작하기 때문에 windows Hadoop을 설치해야 한다. 아래 미리 Windows용으로 빌드된 Hadoop을 설치한다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><a href="https://github.com/karthikj1/Hadoop-2.7.1-Windows-64-binaries/releases/download/v2.7.1/hadoop-2.7.1.tar.gz">https://github.com/karthikj1/Hadoop-2.7.1-Windows-64-binaries/releases/download/v2.7.1/hadoop-2.7.1.tar.gz</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>예를 들어, 압축을 풀고 "C:\app\hadoop"으로 이동시킨 후 환경변수를 설정한다. 현재 자바 최신 버전인 JDK 11은 아직 지원되지 않고 문제가 발생하므로 JDK 8 최신 버전을 설정한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">JAVA_HOME=C:\app\Java\8\jdk<br>HADOOP_HOME=C:\app\hadoop<br>ANACONDA_HOME=C:\app\anaconda <br>PYTHONPATH=%ANACONDA_HOME%<br>PYSPARK_PYTHON=%PYTHONPATH%\python.exe<br>PYTHONIOENCODING=UTF-8<br>PATH=%PATH%;%ANACONDA_HOME%;%ANACONDA_HOME%\Library\bin; <br>%HADOOP_HOME%\bin;</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>Anaconda prompt를 열고 환경변수가 잘 설정되었는지 확인하고 pyspark를 설치한다. 현재 최신 PySpark 버전이 2.4.0이지만 dataframe 생성시 exception이 발생하여 2.3.2로 설치한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">echo %JAVA_HOME%<br>echo %HADOOP_HOME%<br>...<br>conda install pyspark=2.3.2</pre>
<!-- /wp:preformatted -->

<!-- wp:heading -->
<h2>IDE</h2>
<!-- /wp:heading -->

<!-- wp:heading {"level":3} -->
<h3>Jupyter Notebook</h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Anaconda prompt에서 아래 명령어를 실행시키면 데몬이 시작되면서 기본 browser가 자동으로 열린다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"bash","className":"brush: bash; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: bash; notranslate">(base) C:\Users\brad>jupyter notebook
[I 22:47:18.529 NotebookApp] JupyterLab extension loaded from C:\app\anaconda\lib\site-packages\jupyterlab
[I 22:47:18.530 NotebookApp] JupyterLab application directory is C:\app\anaconda\share\jupyter\lab
[I 22:47:18.532 NotebookApp] Serving notebooks from local directory: C:\Users\brad
[I 22:47:18.532 NotebookApp] The Jupyter Notebook is running at:
[I 22:47:18.532 NotebookApp] http://localhost:8888/?token=6d3d60c01b9715eb8d3a94fd862f22eefab07192fe7aa1a0</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:paragraph -->
<p>New &gt; Python 3로 새로운 edit 탭을 시작한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":32} -->
<figure class="wp-block-image"><img src="https://classx365.files.wordpress.com/2018/12/image.png" alt="" class="wp-image-32"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Syntax highlight와 괄호 닫기 자동 입력을 지원해준다. Shift+Enter로 작성한 상자 안에 코드를 수행해볼 수 있다. 작성 중간 중간에 File &gt; Save를 해준다. 현재 directory 안에 .ipynb 화일로 저장된다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":33} -->
<figure class="wp-block-image"><img src="https://classx365.files.wordpress.com/2018/12/image-1.png" alt="" class="wp-image-33"/></figure>
<!-- /wp:image -->

<!-- wp:heading {"level":3} -->
<h3>PyCharm</h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>편리한 편집 기능을 제공하는 PyCharm이다. 단축키와 스타일이 IntelliJ와 유사하다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":35} -->
<figure class="wp-block-image"><img src="https://classx365.files.wordpress.com/2018/12/image-2.png" alt="" class="wp-image-35"/></figure>
<!-- /wp:image -->

<!-- wp:heading {"level":4} -->
<h4>Troubleshooting</h4>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Numpy update 후에 PyCharm에서 다음과 같은 오류가 발생한다</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">"DLL Load Failed: The specified procedure could not be found."</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>Windows 환경 변수 PATH에 다음을 추가하고 PyCharm을 restart해보자.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">%ANACONDA_HOME%\Library\bin </pre>
<!-- /wp:preformatted -->

<h1><strong>Anaconda Installation</strong></h1>

<!-- wp:paragraph -->
<p>Windows 10에 Anaconda를 설치하는 방법을 정리한다. 개인적으로 경로 상에 공백 문자가 없는 것을 선호하기 때문에 C:\app\anaconda 디렉토리 아래에 설치해 보도록 하겠다.</p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Downloading Anacond</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>아래 URL로 가서 최신 Anaconda 설치화일을 다운로드한다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>https://www.anaconda.com/download/</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>현재 최신 버전은, https://repo.continuum.io/archive/Anaconda3-2018.12-Windows-x86_64.exe 이다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>설치 창에서 경로를 "C:\app\anaconda"로 수정하고 설치를 완료한다.</p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Setting Environment Variables</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Windows 10의 시스템 환경변수에 다음과 같이 추가한다.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">ANACONDA_HOME=C:\app\anaconda <br>PYTHONPATH=%ANACONDA_HOME%<br>PYTHONIOENCODING=UTF-8<br>PATH=%PATH%;%ANACONDA_HOME%;%ANACONDA_HOME%\Library\bin;</pre>
<!-- /wp:preformatted -->

<!-- wp:heading -->
<h2>Adding Mirror Site</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Anaconda 설치 후 새로 생긴 Anaconda Prompt를 실행한다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":52,"width":239,"height":210} -->
<figure class="wp-block-image is-resized"><img src="https://classx365.files.wordpress.com/2018/12/image-3.png?w=372" alt="" class="wp-image-52" width="239" height="210"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>기본적으로 Anaconda는 아래 채널을 갖고 있는데 속도가 느려서 중국의 미러 채널을 추가해 보도록하겠다.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">default_channels:<br> https://repo.anaconda.com/pkgs/main<br> https://repo.anaconda.com/pkgs/free<br> https://repo.anaconda.com/pkgs/r<br> https://repo.anaconda.com/pkgs/pro<br> https://repo.anaconda.com/pkgs/msys2 </pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>다음 URL을 참조하여 채널을 추가한다. </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>https://mirrors.tuna.tsinghua.edu.cn/help/anaconda/</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config --set show_channel_urls yes
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/pytorch/</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>테스트 삼아 numpy를 설치해본다.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>conda install numpy</code></pre>
<!-- /wp:code -->

<!-- wp:heading -->
<h2>Updating Conda Package</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>설치되어 있는 Anaconda package들을 살펴보자.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>conda list</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>Package들을 update해본다.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>conda update --all</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>캐쉬된 잔여물들을 삭제한다.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>conda clean --all</code></pre>
<!-- /wp:code -->

<!-- wp:heading -->
<h2>Troubleshooting</h2>
<!-- /wp:heading -->

<!-- wp:heading {"level":3} -->
<h3>Unknown Encoding</h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>만일 conda update 중 아래와 같은 메시지를 만나게 된다면 Windows 10에서 기본 문자셋을 UTF-8로 설정했기 때문일 것이다. 환경변수 "PYTHONIOENCODING=UTF8"으로 설정했음에도 불구하고 6500이라는 숫자로 변경되었다.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">Preparing transaction: done<br> Verifying transaction: done<br> Executing transaction: done<br> Fatal Python error: init_sys_streams: can't initialize sys standard streams<br> LookupError: unknown encoding: 6500</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>Anaconda Prompt를 닫고 "C:\app\anaconda\Scripts\activate.bat" 끝에 아래 라인을 강제로 추가한다.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">set PYTHONIOENCODING=UTF-8</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>다시 Anaconda Prompt를 열고 encoding을 확인한다.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>(base) C:\Users\brad&gt;echo %PYTHONIOENCODING%
UTF-8

(base) C:\Users\brad&gt;conda update conda
Solving environment: /</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->

<h1><strong>Spark SQL</strong></h1>

<!-- wp:paragraph -->
<p>긴 설명은 생략하고 직관적으로 이해할 수 있도록 Python 샘플 코드를 통해서 레퍼런스처럼 참조할 수 있도록 정리한다. Ctrl+F로 검색하기 쉽도록 외래어는 가능하면 알파벳으로 표시할 것이다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Spark SQL Session</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Spark에서 DataFrame을 생성하고 SQL을 수행하기 위해서&nbsp;<br><code>pyspark.sql.SparkSession</code> 인스턴스를 만든다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName("MyApp").getOrCreate()</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>File I/O</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>대부분의 경우 대량의 입력 데이터를 수기로 입력할 수는 없고 다른 data source로부터 받아야 한다. 가장 기본이 되는 방식으로 하나의 File을 읽어서 DataFrame으로 변환해 보자.<br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>테스트를 위하여 다음 링크에서 붓꽃 데이터를 다운로해서 저장한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p> <a href="https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data">https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data</a><br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>아래 예제는 Pyspark v2.4.0 기준이고, 이하 모든 예제는 사용 버전에 따라 코드가 달라질 수 있다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># spark = SparkSession.builder.getOrCreate()
irisDf = spark.read.csv('data/iris.data', sep=",", header="false", inferSchema="true")</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:list -->
<ul><li>sep: 구분자. 기본은 콤마(,)</li><li>header: csv에 헤더가 있는지 여부</li><li>inferSchema: 컬럼 내용을 보고 data type을 유추해서 결정해 준다. Default인 "false"로 하면 string type으로 인식한다.<br></li></ul>
<!-- /wp:list -->

<!-- wp:heading -->
<h2>DataFrames</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>방금 CSV file을 읽어서 dataframe 한 개로 전환하였다. Dataframe의 정확한 정의는 따로 있겠지만 이해하기 쉽도록 schema를 가진 RDD라고 생각하자. DB table처럼 schema가 있기 때문에 SQL로 조작이 가능해진다. 이제 dataframe에 대한 기초 조작법을 정리해 보자. <br></p>
<!-- /wp:paragraph -->

<!-- wp:heading {"level":4} -->
<h4>Schema</h4>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">>>> df.columns
['_c0', '_c1', '_c2', '_c3', '_c4']
>>> df.dtypes
[('_c0', 'double'), ('_c1', 'double'), ('_c2', 'double'), ('_c3', 'double'), ('_c4', 'string')]
>>> df.printSchema()
root
 |-- _c0: double (nullable = true)
 |-- _c1: double (nullable = true)
 |-- _c2: double (nullable = true)
 |-- _c3: double (nullable = true)
 |-- _c4: string (nullable = true)</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Print</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>DataFrame을 화면에 출력하는 기능이다.&nbsp; 앞에 chain이 얼마나 길든 상관없이 최종 결과 DataFrame을 화면에 출력한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># Print a DataFrame in table format. All the belows print the same table.
irisDf.show(3)
irisDf.select('*').show(3)
irisDf.limit(3).show(3)
# Display
+---+---+---+---+-----------+
|_c0|_c1|_c2|_c3|        _c4|
+---+---+---+---+-----------+
|5.1|3.5|1.4|0.2|Iris-setosa|
|4.9|3.0|1.4|0.2|Iris-setosa|
|4.7|3.2|1.3|0.2|Iris-setosa|
+---+---+---+---+-----------+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Query</h2>
<!-- /wp:heading -->

<!-- wp:heading {"level":3} -->
<h3>First N<br></h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>전체 row 또는 앞 n 개 row를 질의할 수 있는 몇가지 방법들을 제공한다. 테스트 결과 아래 구문들은 모두 동일한 Row 객체 리스트를 반환한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># Whatever methods below return the same list of Row object.
irisDf.head(3)
irisDf.take(3)
irisDf.collect()[:3]
irisDf.limit(3).collect()
irisDf.select('*').limit(3).collect()
# Result
[Row(_c0=5.1, _c1=3.5, _c2=1.4, _c3=0.2, _c4='Iris-setosa'), Row(_c0=4.9, _c1=3.0, _c2=1.4, _c3=0.2, _c4='Iris-setosa'), Row(_c0=4.7, _c1=3.2, _c2=1.3, _c3=0.2, _c4='Iris-setosa')]</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:list -->
<ul><li>take(n): 앞의 n개 row를 fetch<br></li><li>show(n): 앞의 n개 row를 print<br></li></ul>
<!-- /wp:list -->

<!-- wp:heading {"level":3} -->
<h3>API Queries<br></h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>SQL과 유사한 API로 질의한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">>>> df.select('_c4').filter('_c0 > 5.0 and _c2 &lt; 1.5').show(3)
+-----------+
|        _c4|
+-----------+
|Iris-setosa|
|Iris-setosa|
|Iris-setosa|
+-----------+
>>> df.groupby('_c4').agg({'_c1':'max', '_c2':'count'}).show()
+---------------+--------+----------+
|            _c4|max(_c1)|count(_c2)|
+---------------+--------+----------+
| Iris-virginica|     3.8|        50|
|    Iris-setosa|     4.4|        50|
|Iris-versicolor|     3.4|        50|
+---------------+--------+----------+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:list -->
<ul><li>filter(): SQL의 WHERE에 해당한다.</li><li>groupby(): SQL과 달리 SELECT가 없고 group by column을 명시하지 않는다.</li><li>agg(): 'column명':'항수명' 형태로 aggregation columns를 표현한다.<br></li><li>show(): 앞의 fiter, groupby, agg 등은 transformations여서 실제 수행은 하지 않고 설계만 하는 것이고, action인 show()를 불러야 실제로 조회가 시작된다.<br></li></ul>
<!-- /wp:list -->

<!-- wp:heading -->
<h2>Columns</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>
지금까지는 windows command 창에서 수행하였는데 코딩이 수월한 PyCharm으로 옮기겠다.

</p>
<!-- /wp:paragraph -->

<!-- wp:heading {"level":3} -->
<h3>Column Functions<br></h3>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>&nbsp;Column를 전문적으로 다루는 module이 따로 있다. Alias, aggregation 등 편리한 기능들이 있다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">from pyspark.sql import functions as F
df.groupby('_c4').agg(F.max('_c1').alias('_c1_max'), F.count('_c2').alias('_c2_count')).show()
+---------------+-------+---------+
|            _c4|_c1_max|_c2_count|
+---------------+-------+---------+
| Iris-virginica|    3.8|       50|
|    Iris-setosa|    4.4|       50|
|Iris-versicolor|    3.4|       50|
+---------------+-------+---------+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading {"level":3} -->
<h3>Add Column<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">df.withColumn('_c5', F.col('_c1')*2).show(3)
+---+---+---+---+-----------+---+
|_c0|_c1|_c2|_c3|        _c4|_c5|
+---+---+---+---+-----------+---+
|5.1|3.5|1.4|0.2|Iris-setosa|7.0|
|4.9|3.0|1.4|0.2|Iris-setosa|6.0|
|4.7|3.2|1.3|0.2|Iris-setosa|6.4|
+---+---+---+---+-----------+---+
df.show(3)
+---+---+---+---+-----------+
|_c0|_c1|_c2|_c3|        _c4|
+---+---+---+---+-----------+
|5.1|3.5|1.4|0.2|Iris-setosa|
|4.9|3.0|1.4|0.2|Iris-setosa|
|4.7|3.2|1.3|0.2|Iris-setosa|
+---+---+---+---+-----------+
df2 = df.withColumn('_c5', F.col('_c1')*2)
df2.show(3)
+---+---+---+---+-----------+---+
|_c0|_c1|_c2|_c3|        _c4|_c5|
+---+---+---+---+-----------+---+
|5.1|3.5|1.4|0.2|Iris-setosa|7.0|
|4.9|3.0|1.4|0.2|Iris-setosa|6.0|
|4.7|3.2|1.3|0.2|Iris-setosa|6.4|
+---+---+---+---+-----------+---+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:list -->
<ul><li>withColumn(): 새로운 column을 마지막에 추가</li><li>df.show(3): Spark RDD나 DataFrame은 mutable 자료구조라 in-place update가 안 된다. 다른 변수에 assign해서 변경을 유지해 나간다.<br></li></ul>
<!-- /wp:list -->

<!-- wp:heading {"level":3} -->
<h3>Update Column</h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">df.withColumn('_c4', df['_c4'].substr(1,4)).show(3)
+---+---+---+---+----+
|_c0|_c1|_c2|_c3| _c4|
+---+---+---+---+----+
|5.1|3.5|1.4|0.2|Iris|
|4.9|3.0|1.4|0.2|Iris|
|4.7|3.2|1.3|0.2|Iris|
+---+---+---+---+----+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading {"level":3} -->
<h3>Type Casting<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">df.select(F.col('_c0').cast('integer')).show(3)
+---+
|_c0|
+---+
|  5|
|  4|
|  4|
+---+</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Join</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>DataFrame class의 member method인 join()은 INNER, LEFT OUTER, RIGHT OUTER, FULL OUTER, CROSS 등의 join 기능을 제공한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># Create python lists.
dept = [(1,'hr'), (2,'dev'), (3,'special_force')]
emp = [(11, 'john', 1), (12, 'james', 2), (13, 'jennis', 1), (14, 'brad', 2), (15, 'doctor-x', 4)]
# Create RDDs.
deptRdd = sc.parallelize(dept)
empRdd = sc.parallelize(emp)
# Create DataFrames.
deptDf = spark.createDataFrame(dept, ['dept_id', 'dept_name'])
empDf = spark.createDataFrame(emp, ['emp_id', 'emp_name', 'dept_id'])
# Join two DataFrames.
joinDf = deptDf.join(empDf, on='dept_id')
joinDf.show()
+-------+---------+------+--------+
|dept_id|dept_name|emp_id|emp_name|
+-------+---------+------+--------+
|      1|       hr|    11|    john|
|      1|       hr|    13|  jennis|
|      2|      dev|    12|   james|
|      2|      dev|    14|    brad|
+-------+---------+------+--------+
deptDf.join(empDf, deptDf.dept_id == empDf.dept_id, 'left_outer').show()
+-------+-------------+------+--------+
|dept_id|    dept_name|emp_id|emp_name|
+-------+-------------+------+--------+
|      1|           hr|    11|    john|
|      1|           hr|    13|  jennis|
|      3|special_force|  null|    null|
|      2|          dev|    12|   james|
|      2|          dev|    14|    brad|
+-------+-------------+------+--------+
deptDf.join(empDf, deptDf['dept_id'] == empDf['dept_id'], 'full_outer').show()
+-------+-------------+------+--------+
|dept_id|    dept_name|emp_id|emp_name|
+-------+-------------+------+--------+
|      1|           hr|    11|    john|
|      1|           hr|    13|  jennis|
|      3|special_force|  null|    null|
|      2|          dev|    12|   james|
|      2|          dev|    14|    brad|
|      4|         null|    15|doctor-x|
+-------+-------------+------+--------+</pre>
<!-- /wp:syntaxhighlighter/code -->

<h1><strong>Spark RDD</strong></h1>

<!-- wp:paragraph -->
<p>Python sample code를 예로 해서 RDD를 어떻게 다루는지 기본 조작 방법들을 정리해 보자.<br></p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>RDD</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>RDD는 <em>Resilient Distributed Dataset</em>의 약자인데 이름에 유추할 수 있듯이 fault-tolerant하고 분산되어 병렬 처리되는 자료구조이다. 우리는 보통 S3, HDFS, DB에 있는 source data를 읽어서 RDD 형태로 로드하게 된다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Spark SQL이 테이블 형태의 스키마를 가지고 SQL로 조작하기 위한 자료구조라면, RDD는 map-reduce로 조작하기 위한 자료구조이다. 양방향으로 치환이 가능하다. 따라서 주로 python tuple 형태 (k, v)로 data를 만들어 놓고, map/reduce 함수로 조작하게 된다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>하나의 RDD는 논리적인 하부구조인 하나 이상의 partition으로 이루어진다. <br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>RDD에 대한 기본 조작을 python 예제로 살펴보자. 관련 method가 많으므로 자주 사용되는 몇가지만 살펴본다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Cluster Structure<br></h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p> Driver program은 우리들이 작성한 Spark code이고 PC client에서 실행할 수도 있고 Spark cluster node 안에서 실행시킬 수도 있다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>지금부터의 클러스터 내용은 정확하지 않을 수 있는데 나중에 수정하도록 하겠다. Driver가 data를 load해서 cluster의 각 node에 분산시켰다고 하자. Driver는 cluster node 중에 지정된 app master에게 RDD 조작 명령을 내린다. App master는 RDD를 보관하고 있는 각 executor node에게 다시 명령을 전달한다.&nbsp; <br></p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Spark Context</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>Spark cluster에 접속하기 위해서 SparkContext를 생성한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName('MyApp').setMaster('local[4]')
sc = SparkContext(conf=conf)</pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Creating RDD<br></h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>아래에서 생성한 RDD는 이후 example에서도 계속 사용된다. <br></p>
<!-- /wp:paragraph -->

<!-- wp:heading {"level":3} -->
<h3>From Driver Program<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># Small printer
import pprint
def pp(o):
    pprint.pprint(o)
    pprint.pprint(type(o))

# parallelize: Get a Python data and distribute on Spark cluster nodes.
personList = [('David',101),('Bob',102),('Charles',103),('Brad',104)]
personRdd = sc.parallelize(personList)
pp(personRdd)
ParallelCollectionRDD[13] at parallelize at PythonRDD.scala:194
&lt;class 'pyspark.rdd.RDD'></pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading {"level":3} -->
<h3>From External Data Source<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate">filePath = 'data/hollins.dat.gz'
# textFile: Open a file and return a RDD.
fileRdd = sc.textFile(filePath)
pp(fileRdd.count())
29888
&lt;class 'int'></pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Listing Elements</h2>
<!-- /wp:heading -->

<!-- wp:heading {"level":3} -->
<h3>All<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># collect: Return a python list from a RDD.
pp(personRdd.collect())
[('David', 101), ('Bob', 102), ('Charles', 103), ('Brad', 104)]
&lt;class 'list'></pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading {"level":3} -->
<h3>First N<br></h3>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># first: Return the 1st element of list.
pp(personRdd.first())
('David', 101)
&lt;class 'tuple'>

# take: Return the first n elements of list.
pp(personRdd.take(2))
[('David', 101), ('Bob', 102)]
&lt;class 'list'>

# top: Return the largest n elements of list.
pp(personRdd.top(2))
[('David', 101), ('Charles', 103)]
&lt;class 'list'></pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Key-Value</h2>
<!-- /wp:heading -->

<!-- wp:syntaxhighlighter/code {"language":"python","className":"brush: python; notranslate"} -->
<pre class="wp-block-syntaxhighlighter-code brush: python; notranslate"># lookup: like get() in most cache systems
pp(personRdd.lookup('Charles'))
[103]
&lt;class 'list'></pre>
<!-- /wp:syntaxhighlighter/code -->

<!-- wp:heading -->
<h2>Transformation Operations<br></h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>map/groupby/join 등과 같이 RDD를 변형시키는 연산들이며 연산 결과가 다시 RDD이므로 연산들을 chainning할 수 있다. Lazy approach라서 transformation은 여러 번을 적용하더라도 실제로 연산이 수행되지는 않는다. Action 연산을 적용할 때 한꺼번에 수행된다. Transformation/Action 연산 리스트는 아래 spark guide page에 나열되어 있다.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>https://spark.apache.org/docs/latest/rdd-programming-guide.html<br></p>
<!-- /wp:paragraph -->

<!-- wp:heading -->
<h2>Action Operations</h2>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>reduce/count/collect와 같이 RDD를 python scalar나 list 등으로 최종 결과를 얻어낼 때 사용한다. 그 앞에 적용한 transformation 연산들을 있다면 실제로 순서대로 수행하고 마지막에 action을 수행한다.<br></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p></p>
<!-- /wp:paragraph -->