빌드서버에서 빌드 후 배포(배치 서버로)

build는 shell을 통해 처리 하였다. (음..뭔가 ansible style은 아니다..ㅠㅠ 책을 사서 좀 더 파봐야겠다.)

- name: Execute sbt build
  shell: |
    cd {{ build_home }}/{{ build_id }}
    pwd
    ./sbt.sh   

sbt.sh에서는 sbt를 통해 명령어로 처리! sbt가 export가 잘안되어서 그냥 절대경로로 처리!

deploy는 간단히 copy해주는 형식으로 처리 하였다. 즉, local jar를 deploy할 서버로 카피!

---
- hosts : spark
  serial : 1
  tasks :
  - name : Make data directory
    file : dest="{{ base_data }}" owner={{ user_name }} group={{ group_name }} state=directory

  - name : Remove old file
    file : path="{{ base_service }}/{{ service_name }}/bin/{{ file_name }}.{{ file_type }}" state=absent

  - name : Deploy file
    copy : src="{{ data_build_home }}/{{ build_id }}/target/scala-2.10/{{ file_name }}.{{ file_type }}"
           dest="{{ base_service }}/{{ service_name }}/bin/{{ file_name }}.{{ file_type }}" 

배치 서버에서 스파크 수행

Spark Submit

애월 - 지금이순간 카페

스칼라 알못 스파크 알못이라..이번에 작업한 내용이 있는데 삽질을 많이 했다...ㅋㅋ 
생각나는 것들을 적어보자.
HDFS를 가져와 RDD에 저장! sc.textFile을 통해서 HDFS를 가져온다.
sc.textFile("hdfs path")
그런데 그냥 가져오면 소용이 없다. 그러므로 map을 통해서 필요한 친구들만 가져온다. 또한 filter를 통해 데이터를 줄여준다.

ex) hdfs의 포맷이 Json이라서 Gson을 사용. filter를 통해 데이터를 줄여준다.

그리고 여러개의(multi) RDD들이 있었고 이 친구들을 한번에 분석을 해야했다. 그래서 union을 사용.
ex) val result  = rddRaw0.union(rddRaw1)

그런뒤 reduceByKey를 하여 word count를 해준다.
rddRaw.map(line => ((line.document.id,device),1)).reduceByKey{ case (x, y) => x + y}

이제 분석되어진 RDD 친구를 List화 하여 JDBC를 통해 DB에 넣는다.
여기에서 중요한 RDD to List는 아래의 내용이다.
val list: List[((String, String), Int)] = result.collect().toList 
그리고 map으로 원하는 형태를 만들어준 뒤,  InsertToDb를 해주면 되는데 여기에서 sirialization 문제가 발생한다.

Serialize 해결은 아래와 같이 하였다.

Serializable은 해결방법보다 이녀석이 뭔지 정확히 알필요가 있으니 따로 포스팅을 하겠습니다.

자 이제 InsertToDb에서는 connection을 맺고 db작업을 해주면 된다.
여기에서의 팁은 아래에 보시면 getConnection이 아니라 getPoolConnection()인 것을 알수가 있다.

1) get connection
    val connection = OutputJdbcTask.getPoolConnection()

pool 사용을 위해 build.sbt에 아래를 추가

소스는 아래와 같다.

val env = MainConfig.env
val driver = ConfigFactory.load().getString(s"$env.driver")
val url = ConfigFactory.load().getString(s"$env.url")
val username = ConfigFactory.load().getString(s"$env.username")
val password = ConfigFactory.load().getString(s"$env.password")

private val cpds: ComboPooledDataSource = new ComboPooledDataSource()
cpds.setDriverClass(driver)
cpds.setJdbcUrl(url)
cpds.setUser(username)
cpds.setPassword(password)
cpds.setMinPoolSize(5)
cpds.setAcquireIncrement(5)
cpds.setMaxPoolSize(35)

def getConnection(){
  try {
    Class.forName(driver)
    DriverManager.getConnection(url, username, password)
  } catch {
    case e : Exception => e.printStackTrace
  }
}

def getPoolConnection() = {
  cpds.getConnection
}

마지막으로 DB Upsert하는 부분 소스는 아래와 같다.

try {
  val connection = OutputJdbcTask.getPoolConnection()

  val insertSql =
    """
      |insert into 테이블명 (dsid, reg_dttm, click_count, device, create_date )
      |values (?,?,?,?,?)
      |ON DUPLICATE KEY UPDATE dsid = ?, reg_dttm =?, click_count= ?, device =?, create_date =?
    """.stripMargin

  val insertStmt: PreparedStatement = connection.prepareStatement(insertSql)
  val regDttm = MainConfig.currentTimeAsString
  val createDate = MainConfig.date
  var itemCnt = 1
  val chunk = 10000

  targetList.foreach {
    t =>
      val dsid = t._1
      val device = t._2
      val clickCnt = t._3

      insertStmt.setString(1, dsid)
      insertStmt.setString(2, regDttm)
      insertStmt.setInt(3, clickCnt)
      insertStmt.setString(4, device)
      insertStmt.setString(5, createDate)
      insertStmt.setString(6, dsid)
      insertStmt.setString(7, regDttm)
      insertStmt.setInt(8, clickCnt)
      insertStmt.setString(9, device)
      insertStmt.setString(10, createDate)
      insertStmt.addBatch()
      insertStmt.clearParameters()
      if (itemCnt % chunk == 0) {
        insertStmt.executeBatch()
      }
      itemCnt = itemCnt + 1

  }
  insertStmt.executeBatch()
  insertStmt.close()
  connection.close()
} catch {
  case e: Exception => {
    e.printStackTrace()
  }
}

참고 사이트)

https://stackoverflow.com/questions/40892800/spark-rdd-to-list

https://alvinalexander.com/scala/how-to-use-serialization-in-scala-serializable-trait

1. hdfs 연동. 즉, 다른쪽에 저장되어있는 친구를 불러온다(raw data)
2. Spark 분석(map-reduce)
3. Mysql에 저장.
1. scala application.conf의 정보 읽어와서 JDBC 접속 후 INSERT 등 수행.

https://spark.apache.org/docs/latest/rdd-programming-guide.html

spark rdd

Overview

At a high level, every Spark application consists of a driver program that runs the user’s main function and executes various parallel operations on a cluster. 

The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.

A second abstraction in Spark is shared variables that can be used in parallel operations. 
By default, when Spark runs a function in parallel as a set of tasks on different nodes, it ships a copy of each variable used in the function to each task. Sometimes, a variable needs to be shared across tasks, or between tasks and the driver program. Spark supports two types of shared variables: broadcast variables, which can be used to cache a value in memory on all nodes, and accumulators, which are variables that are only “added” to, such as counters and sums.

This guide shows each of these features in each of Spark’s supported languages. It is easiest to follow along with if you launch Spark’s interactive shell – either bin/spark-shell for the Scala shell or bin/pyspark for the Python one.

2. flatMap[U](f: (T) ⇒ TraversableOnce[U])(implicit arg0: ClassTag[U]): RDD[U]

   

   Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.

map[U](f: (T) ⇒ U)(implicit arg0: ClassTag[U]): RDD[U]

Return a new RDD by applying a function to all elements of this RDD.

2016/10/18 - [BigDATA/spark] - 1탄. SPARK를 설치해보자~

1탄. 단일모드 

2탄. 클러스터 모드 

3탄. 기타 유용 셋팅(스파크 관련)

우선 분산 클러스터로 셋팅하기전에! 요것만큼은 알고가자는 의미에서 살짝 정리해본다.

1. Spark 구조

(펌: https://www.google.co.kr/search?q=spark+%EA%B5%AC%EC%A1%B0&newwindow=1&biw=1598&bih=976&source=lnms&tbm=isch&sa=X&sqi=2&ved=0ahUKEwi1y4ut0ObPAhWqiVQKHWFaDgcQ_AUIBigB&dpr=1#imgrc=EdvQ87Vu0XWkMM%3A)

2. Spark RDD..꼭 알아야하나?

알아야한다~코딩하려면 뭔지는 알고 코딩하자!

또한 스파크의 핵심기능으로써 분산 되어있는 변경불가능한 객체의 모음이라고 생각하자.

2가지 타입으로 구분

   1) transformation 2) action 으로 구분되어진다.

간단히 말하면 1)은 기존 RDD를 new RDD로 리턴 하는 것(ex. filter(), map() 등)

2)는 기존 RDD를 계산하여 저장하거나 다른 타입으로 리턴하는 것.

Spark는 RDD의 내용을 메모리에 클러스터의 머신들에 나뉘어서 저장 -> action에서 재사용!

또한 실행구조는 분산모드일 경우!

마스터 / 슬레이브의 구조!

여기에서 마스터는 드라이버(driver)라고 불림. 슬레이브는 익스큐터(executor) => 이것을 Spark Application이라고 함!

이녀석들은 서로 독립된 java process로 돌아감.

용어중에 Task는 스파크 작업 계층에서 최소 개체라고 보면 된다.

3. 실행 흐름

1.  사용자 프로그램을 스파크에 제출! -> spark-submit.sh 
2.  드라이버 실행 : 익스큐터 실행을 위한 리소스를 클러스터 매니저에 요청.
3. 클러스터 매니저는 익스큐터를 실행.
4. 드라이버는 main() 메소드를 호출! 이때 작업 내역을 단위 작업 형태(task)로 나눠 익스큐터에게 보낸다.
5. 익스큐터는 task를 실행.
6. 드라이버 : main()이 끝이거나 SparkContext.stop() 호출 시 익스큐터들을 중지! 
   클러스터 매니저에 사용했던 자원을 반환!

해당 포스팅은 총 3부로 구성될 예정 입니다.

1탄. 단일모드 

2탄. 클러스터 모드

3탄. 기타 유용 셋팅(스파크 관련)

1탄. SPARK를 설치해보자~

Apache Spark 설치!

버전은 1.6.1 / hadoop 2.6으로 해서 다운받아보겠습니다. (현업에서 사용중인게 요거라서 요걸로!)

Step 1. 아래의 링크를 통해 스파크를 다운 받아보자!

http://spark.apache.org/downloads.html

Step 2. down을 받아서 원하는 서버에서 압축을 풀어줍니다.

압축을 해제하고 내용을 보면 아래와 같습니다.

Step 3. Spark는 대화형 쉘들을 제공 합니다.

파이썬과 스칼라가 있는데요 

즉석 데이터를 분석하기에 좋습니다.

실행은?

1) 파이썬 쉘  

ㄴ bin directory에 가서 ./pyspark 를 실행시키면 아래와 같이 수행되어집니다.

2) 스칼라 쉘

ㄴ bin/spark-shell 를 수행!

간단히 테스트를 해보자.

내용은 스파크 퀵 스타트를 통해 해보자!

http://spark.apache.org/docs/latest/quick-start.html

따라해보기!

Step 4. Log 설정을 해보자.

Spark에서의 로그는 아래와 같이 설정할 수가 있다.

아래의 템플릿 중에 log4j.properties.template를 복사해서 log4j.properties를 만들면 된다.

많이 사용되어지는 친구이기 때문에 잘 아실거라 믿는다.

내용은!? 아래처럼 INFO로 설정이 되어있어서 spark-shell를 수행시키면 많은 정보들이 나오게 된다. 
INFO->WARN으로 변경을 한다면 적은 내용의 정보가 보인다.

- 1탄 끝~ -