Spark
인-메모리 기반의 클러스터 컴퓨팅 프레임워크인 Spark를 정리 합니다.
- 홈페이지 : http://spark-project.org/, https://spark.apache.org/
- 다운로드 :
- 라이선스 :
- 플랫폼 : Scala
- API : Java, Scala, Python, R
Spark 개요
Apach Spark UC 버클리 대학의 AMPLab에서 내놓은 대용량 분산 처리 및 분석용 오픈소스이다. 2014년 2월부터 아파치 재단의 톱 프로젝트가 되었다.
- 대화형 질의 분석기(Shark), 대용량 그래프 처리 및 분석기(Bagel), 실시간 분석기(Spark Streaming) 등을 함께 제공
Spark 구성
Scala 설치
cd /install wget https://downloads.lightbend.com/scala/2.12.3/scala-2.12.3.tgz cd /appl tar -xvzf /install/scala-2.12.3.tgz mv scala-2.12.3.tgz scala # export PATH=${PATH}:/appl/scala/bin
Spark 설치
Spark 설치
cd /install wget http://apache.mirror.cdnetworks.com/spark/spark-2.3.0/spark-2.3.0-bin-hadoop2.7.tgz cd /appl tar -xvzf /install/spark-2.3.0-bin-hadoop2.7.tgz mv spark-2.3.0-bin-hadoop2.7 spark cd /appl/spark cd conf cp spark-env.sh.template spark-env.sh cp log4j.properties.template log4j.properties vi spark-env.sh vi log4j.properties log4j.rootCategory=WARN, console # cd /appl/spark # sbin/start-master.sh # sbin/start-slave.sh spark://localhost:7077 # bin/pyspark -master spark://localhost:7077 cd /appl/spark sbin/start-all.sh bin/pyspark # bin/spark-shell
- Pyspark : http://localhost:4040/
- Spark : http://localhost:8080/
폴더 구성
- R/
- bin/
- conf/
- data/
- examples/
- jars/
- kubernetes/
- licenses/
- python/
- sbin/
- yarn/
K-ICT 교육
Spark 설치
cd ~ mkdir install cd ~/install wget http://apache.mirror.cdnetworks.com/spark/spark-2.3.0/spark-2.3.0-bin-hadoop2.7.tgz cd ~ tar -xvzf /install/spark-2.3.0-bin-hadoop2.7.tgz mv spark-2.3.0-bin-hadoop2.7 spark cd ~/spark cd conf cp spark-env.sh.template spark-env.sh cp log4j.properties.template log4j.properties vi spark-env.sh export LANG=ko_KR.UTF-8 export JAVA_HOME=/usr/lib/jvm/jre-1.7.0-openjdk.x86_64 export PATH=$PATH:$JAVA_HOME export HADOOP_INSTALL=/usr/local/hadoop export HADOOP_MAPRED_HOME=$HADOOP_INSTALL export HADOOP_COMMON_HOME=$HADOOP_INSTALL export HADOOP_HDFS_HOME=$HADOOP_INSTALL export YARN_HOME=$HADOOP_INSTALL export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_INSTALL/lib/native export PATH=$PATH:$HADOOP_INSTALL/sbin export PATH=$PATH:$HADOOP_INSTALL/bin export SPARK_DIST_CLASSPATH=$(hadoop classpath) vi log4j.properties log4j.rootCategory=WARN, console cd ~/spark sbin/start-all.sh bin/pyspark
- Hadoop Resource Manager : http://localhost:8088/
- Hadoop Node Manager : http://localhost:8042/
- Pyspark : http://localhost:4040/
- Spark : http://localhost:8080/
DataFrame으로 로드
RDD (Resilient Distributed Dataset)
JSON 파일 로드
df = sqlContext.read.json("file:///home/eduuser/spark/examples/src/main/resources/people.json") df = sqlContext.read.load("file:///home/eduuser/spark/examples/src/main/resources/people.json", format="json")
Text 파일 로드
from pyspark.sql import Row lines = sc.textFile("file:///home/eduuser/spark/examples/src/main/resources/people.txt") parts = lines.map(lambda l: l.split(",")) people = parts.map(lambda p: Row(name=p[0], age=int(p[1]))) df = sqlContext.createDataFrame(people)
Text 파일 로드 with Schema 지정
from pyspark.sql.types import * lines = sc.textFile("file:///home/eduuser/spark/examples/src/main/resources/people.txt") parts = lines.map(lambda l: l.split(",")) people = parts.map(lambda p: (p[0], p[1].strip())) schemaString = "name age" fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()] schema = StructType(fields) schemaPeople = sqlContext.createDataFrame(people, schema)
Parquet 데이터 로드
df = sqlContext.read.load("file:///home/eduuser/spark/examples/src/main/resources/users.parquet")
DataFrame을 저장
df.select("name", "favorite_color").write.save("file:///home/eduuser/namesAndFavColors.parquet") # df = sqlContext.read.load("file:///home/eduuser/namesAndFavColors.parquet") df.select("name", "age").write.save("file:///home/eduuser/namesAndAges.parquet", format="parquet")
DataFrame을 TempTable로 지정
df.registerTempTable("people") teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") teenagers.show()
pyspark Sample 1
help(sqlContext) q df = sqlContext.read.json("file:///home/eduuser/spark/examples/src/main/resources/people.json") df.show() df.printSchema() df.select("name").show() df.select(df['name'], df['age'] + 1).show() df.filter(df['age'] > 21).show() df.groupBy("age").count().show() quit()
pyspark Sample 2
from pyspark.sql import Row lines = sc.textFile("file:///home/eduuser/spark/examples/src/main/resources/people.txt") parts = lines.map(lambda l: l.split(",")) people = parts.map(lambda p: Row(name=p[0], age=int(p[1]))) schemaPeople = sqlContext.createDataFrame(people) schemaPeople.registerTempTable("people") teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") teenagers.show() # teenNames = teenagers.map(lambda p: "Name: " + p.name) # for teenName in teenNames.collect(): # print(teenName) quit()
pyspark Sample 3
from pyspark.sql.types import * lines = sc.textFile("file:///home/eduuser/spark/examples/src/main/resources/people.txt") parts = lines.map(lambda l: l.split(",")) people = parts.map(lambda p: (p[0], p[1].strip())) schemaString = "name age" fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()] schema = StructType(fields) schemaPeople = sqlContext.createDataFrame(people, schema) schemaPeople.registerTempTable("people") teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") teenagers.show() # teenNames = teenagers.map(lambda p: "Name: " + p.name) # for teenName in teenNames.collect(): # print(teenName) quit()
pyspark Sample 4
#--- 데이터 로드/저장 df = sqlContext.read.load("file:///home/eduuser/spark/examples/src/main/resources/users.parquet") #--- 저장시 /home/eduuser/namesAndFavColors.parquet/ 폴더가 생성 됩니다. df.select("name", "favorite_color").write.save("file:///home/eduuser/namesAndFavColors.parquet") df = sqlContext.read.load("file:///home/eduuser/namesAndFavColors.parquet") df = sqlContext.read.load("file:///home/eduuser/spark/examples/src/main/resources/people.json", format="json") df.select("name", "age").write.save("file:///home/eduuser/namesAndAges.parquet", format="parquet")
DataSet 다운로드
cd ~ cd nia_kbig [eduuser@localhost nia_kbig]$ ./datasetDownload.sh 다운로드받을 데이터셋 코드를 입력하세요. 8h7k4 ka988 z24nt cd view/basic ls -alF -rwxr-xr-x. 1 eduuser eduuser 216 2014-12-14 08:46 01.move_data_file.sh* -rwxr-xr-x. 1 eduuser eduuser 283 2014-12-04 22:22 01.move_data_file.sh~* -rwxr-xr-x. 1 eduuser eduuser 270 2014-12-12 16:20 jeju_2010.csv* -rwxr-xr-x. 1 eduuser eduuser 273 2014-12-12 16:16 jeju_2011.csv* -rwxr-xr-x. 1 eduuser eduuser 278 2014-12-12 16:16 jeju_2012.csv* -rwxr-xr-x. 1 eduuser eduuser 1476 2015-01-11 22:59 view_basic_analysis.r* # hdfs dfs -mkdir /user # hdfs dfs -mkdir /user/eduuser/ # hdfs dfs -put jeju* /user/eduuser/
pyspark Sample 5
from pyspark.sql import Row lines = sc.textFile('file:///home/eduuser/nia_kbig/view/basic/jeju_2010.csv') parts = lines.map(lambda l: l.split(',')) jeju2010 = parts.map(lambda p: Row(IN=p[0], OUT=p[1], INCOME=p[2])) schema2010 = sqlContext.createDataFrame(jeju2010) schema2010.registerTempTable("jeju2010") schema2010.show() sqlContext.sql("select * from jeju2010 where INCOME != 'INCOME'").show() jeju2010 = sqlContext.sql("select * from jeju2010 where INCOME != 'INCOME'") jenu2010.show()
pyspark Sample 6
from pyspark.sql.types import * lines = sc.textFile('file:///home/eduuser/nia_kbig/view/basic/jeju_2011.csv') parts = lines.map(lambda l: l.split(',')) jeju2011 = parts.map(lambda p: (p[0].strip(), p[1].strip(), p[2].strip())) schemaString = "IN OUT INCOME" fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()] schema = StructType(fields) schema2011 = sqlContext.createDataFrame(jeju2011, schema) schema2011.registerTempTable('jeju2011') schema2011.show()