[빅데이터] 군집 분석

🎁 본 글은 실무로 '배우는 빅데이터기술' 책을 따라해보고 실행하여보는 과정을 기록한 글이다.

🎁 빅데이터 처리의 전체적인 흐름과 과정을 학습하기 쉬우며 빅데이터에 관심있는 사람들에게 추천한다.

 

 

◼ 임팔라에서 다음을 실행 해 본다. 해당 데이터들을 이용하여 군집 분석을 진행 할 거이다. (Impala에서 실행)

  select * from smartcar_master limit 100;

 

◼ 해당 데이터들을 가공하여 파일을 만들어 주자. (Hive 에디터에서 실행)

insert overwrite local directory '/home/pilot-pjt/mahout-data/clustering/input'
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ' '
select 
  car_number,
  case 
      when (car_capacity < 2000) then '소형'
      when (car_capacity < 3000) then '중형'
      when (car_capacity < 4000) then '대형'
  end as car_capacity,
  case
      when ((2016-car_year) <= 4)  then 'NEW' 
      when ((2016-car_year) <= 8)  then 'NORMAL' 
      else 'OLD'
  end as car_year ,
  car_model,
  sex as owner_sex,
  floor (cast(age as int) * 0.1 ) * 10 as owner_age,
  marriage as owner_marriage,
  job as owner_job,
  region as owner_region
from smartcar_master

 

◼ 위의 명령어가 실행이 안되거나 0%에서 실행이 되지 않고 아래 에러 코드가 나왔다.

Error while processing statement: FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask

이런 사람은 아래 코드를 실행 해준다.

실행 시킬때 주의점은 Ctrl + A 를하여 모든 라인을 드래그 하여 실행 시켜 주어야한다.

set hive.exec.mode.local.auto=true;
insert overwrite local directory '/home/pilot-pjt/mahout-data/clustering/input'
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ' '
select 
  car_number,
  case 
      when (car_capacity < 2000) then '소형'
      when (car_capacity < 3000) then '중형'
      when (car_capacity < 4000) then '대형'
  end as car_capacity,
  case
      when ((2016-car_year) <= 4)  then 'NEW' 
      when ((2016-car_year) <= 8)  then 'NORMAL' 
      else 'OLD'
  end as car_year ,
  car_model,
  sex as owner_sex,
  floor (cast(age as int) * 0.1 ) * 10 as owner_age,
  marriage as owner_marriage,
  job as owner_job,
  region as owner_region
from smartcar_master

 

◼ 실행 후 server02 에서 다음을 실행하여 데이터를 확인한다.

more /home/pilot-pjt/mahout-data/clustering/input/*

 

◼ 데이터 목록은 아래에서 확인 할 수 있다.

ls -al /home/pilot-pjt/mahout-data/clustering/input

 

◼ 이후 폴더를 만들어 준다. 데이터가 나오는 폴더이다.

hdfs dfs -mkdir -p /pilot-pjt/mahout/clustering/input

 

◼ 해당 폴더 위치로 이동.

cd /home/pilot-pjt/mahout-data/clustering/input

 

◼ 파일 이름을 변경하자.

mv 000000_0 smartcar_master.txt

 

hdfs dfs -put smartcar_master.txt /pilot-pjt/mahout/clustering/input

 

◼ Hue 브라우저의 메뉴 > 파일로 이동한다.
/pilot-pjt/mahout/clustering/input 해당 경로의

smartcar_master.txt 파일이 존재 확인 되면 완료 된 것이다.

 

◼ bigdata.smartcar.mahout-1.0.jar 파일을 아래의 경로에 MobaXterm을 사용하여 업로드 한다.

cd /home/pilot-pjt/mahout-data

 

◼ 스마트카 사용자 마스터 텍스트 파일을 시퀀스 파일로 변환한다. (server02 에서 진행)

hadoop jar /home/pilot-pjt/mahout-data/bigdata.smartcar.mahout-1.0.jar com.wikibook.bigdata.smartcar.mahout.TextToSequence /pilot-pjt/mahout/clustering/input/smartcar_master.txt /pilot-pjt/mahout/clustering/output/seq

 

/pilot-pjt/mahout/clustering/output/seq 경로의 파일을 보자
part-m-00000 파일이 존재하는지 확인한다.
◼ 텍스트로 보기를 클릭하여 데이터를 확인 (아래 명령문으로 불 수 도 있다.)

hdfs dfs -text /pilot-pjt/mahout/clustering/output/seq/part-m-00000

 

◼ vector 데이터로 변환처리

mahout seq2sparse -i /pilot-pjt/mahout/clustering/output/seq -o /pilot-pjt/mahout/clustering/output/vec -wt tf -s 5 -md 3 -ng 2 -x 85 --namedVector

 

◼ 다음과 같은 결과가 나올 것이다.

 

 

pilot-pjt/mahout/clustering/output/vec 해당 디렉토리에 df-count,df-vectors 디렉토리가 생성 될 것이다.

 

잠시 캐노피 알고리즘에 대해서 알고가자.

❓ Canopy 군집분석

===========================================================

두계의 임계값 T1, T2를 이용, T1>T2

일련의 점으로 시작하여 무작위로 하나를 제거.

이 점을 포함하는 캐노피를 만들고 점 세트의 나머지 부분을 반복

각 점에서 첫 번째 점으로부터의 거리가 < T1이면 클러스터에 점을 추가

또한 거리가 T2보다 작으면 세트에서 점을 제거

 

T1 (loose threshold)

💨 기준점, 느슨한 거리

T2 (tight threshold)

💨 좁은 거리

===========================================================

 

◼ server02 에 다름 내용 실행

mahout canopy -i /pilot-pjt/mahout/clustering/output/vec/tf-vectors/ -o /pilot-pjt/mahout/clustering/canopy/out -dm org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure -t1 12 -t2 10 -ow

 

◼ 클러스터 수 확인

mahout clusterdump -i /pilot-pjt/mahout/clustering/canopy/out/clusters-*-final

 

◼ 새 notebook 생성
SmartCar-Clustering

//그림 7.96 스파크ML 라이브러리 임포트--------------------------------------------------------------

import org.apache.spark.ml.feature.MinMaxScaler
import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.ml.clustering.KMeansModel
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.ml.evaluation.ClusteringEvaluator

//그림 7.97 스마트카 마스터 데이터셋 로드--------------------------------------------------------------

val ds = spark.read.option("delimiter", " ")
                   .csv("/pilot-pjt/mahout/clustering/input/smartcar_master.txt")
ds.show(5)

//그림 7.98 스마트카 마스터 데이터셋 재구성--------------------------------------------------------------

val dsSmartCar_Master_1 = ds.selectExpr(
                        "cast(_c0 as string) car_number",
                        "cast(_c1 as string) car_capacity",
                        "cast(_c2 as string) car_year",
                        "cast(_c3 as string) car_model",
                        "cast(_c4 as string) sex",
                        "cast(_c5 as string) age",
                        "cast(_c6 as string) marriage",
                        "cast(_c7 as string) job",
                        "cast(_c8 as string) region"
                       )

//그림 7.99 문자형 칼럼을 연속형(숫자형) 칼럼으로 변환 및 생성--------------------------------------------------------------


val dsSmartCar_Master_2 = new StringIndexer().setInputCol("car_capacity").setOutputCol("car_capacity_n")
                                             .fit(dsSmartCar_Master_1).transform(dsSmartCar_Master_1)
val dsSmartCar_Master_3 = new StringIndexer().setInputCol("car_year").setOutputCol("car_year_n")
                                             .fit(dsSmartCar_Master_2).transform(dsSmartCar_Master_2)
val dsSmartCar_Master_4 = new StringIndexer().setInputCol("car_model").setOutputCol("car_model_n")
                                             .fit(dsSmartCar_Master_3).transform(dsSmartCar_Master_3)
val dsSmartCar_Master_5 = new StringIndexer().setInputCol("sex").setOutputCol("sex_n")
                                             .fit(dsSmartCar_Master_4).transform(dsSmartCar_Master_4)
val dsSmartCar_Master_6 = new StringIndexer().setInputCol("age").setOutputCol("age_n")
                                             .fit(dsSmartCar_Master_5).transform(dsSmartCar_Master_5)
val dsSmartCar_Master_7 = new StringIndexer().setInputCol("marriage").setOutputCol("marriage_n")
                                             .fit(dsSmartCar_Master_6).transform(dsSmartCar_Master_6)
val dsSmartCar_Master_8 = new StringIndexer().setInputCol("job").setOutputCol("job_n")
                                             .fit(dsSmartCar_Master_7).transform(dsSmartCar_Master_7)
val dsSmartCar_Master_9 = new StringIndexer().setInputCol("region").setOutputCol("region_n")
                                             .fit(dsSmartCar_Master_8).transform(dsSmartCar_Master_8)


//그림 7.100 클러스터링에 사용할 피처 변수 선택--------------------------------------------------------------

val cols = Array("car_capacity_n", "car_year_n", "car_model_n", "sex_n", "marriage_n")

val dsSmartCar_Master_10 = new VectorAssembler().setInputCols(cols).setOutputCol("features")
                                                .transform(dsSmartCar_Master_9)
val dsSmartCar_Master_11 = new MinMaxScaler().setInputCol("features").setOutputCol("scaledFeatures")
                                             .fit(dsSmartCar_Master_10).transform(dsSmartCar_Master_10)

//그림 7.101 스마트카 마스터 데이터셋 확인 및 학습/검증 데이터 생성----------------------------------------------


val dsSmartCar_Master_12 = dsSmartCar_Master_11.drop("car_capacity").drop("car_year").drop("car_model").drop("sex")
                                               .drop("age").drop("marriage").drop("job").drop("region").drop("features")
                                               .withColumnRenamed("scaledfeatures", "features")
dsSmartCar_Master_12.show(5)                                               

val Array(trainingData, testData) = dsSmartCar_Master_12.randomSplit(Array(0.7, 0.3))

//그림 7.102 스파크ML에서의 K-Means 군집 분석 실행--------------------------------------------------------------


val kmeans = new KMeans()
  .setSeed(1L)
  .setK(200)
  .setFeaturesCol("features")
  .setPredictionCol("prediction")
val kmeansModel = kmeans.fit(dsSmartCar_Master_12)


//그림 7.103 스파크ML에서의 K-Means 군집 결과 확인--------------------------------------------------------------


val transKmeansModel = kmeansModel.transform(dsSmartCar_Master_12)
transKmeansModel.groupBy("prediction").agg(collect_set("car_number").as("car_number")).orderBy("prediction").show(200, false)


//그림 7.104 스파크ML에서의 K-Means 군집 모델 평가 – 실루엣 스코어--------------------------------------------------------------

val evaluator = new ClusteringEvaluator()
val silhouette = evaluator.evaluate(transKmeansModel)

println(s"Silhouette Score = $silhouette")

 

◼ Impala 에디터 에서 작업 (in 뒤의 데이터들은 자신의 데이터를 넣어 주어야 한다.)

select * 
  from smartcar_master
 where car_number in
('I0033', 'P0080', 'V0025', 'A0046', 'Q0019', 'W0020', 'R0051', 'I0081', 'V0039', 'B0015', 'L0051', 'W0030', 'J0083', 'C0069', 'M0043', 'V0095', 'I0045', 'N0005', 'Q0068', 'E0066', 'E0055', 'J0058', 'J0062', 'S0003', 'R0010')

 

◼ 스쿱(sqoop) 사용

◼ hue - 메뉴 - 파일
/user/hive/warehouse/managed_smartcar_symptom_info
위 경로에 있는 파일내용을 postgreSql 데이터베이스로 export  할 것이다.

◼ postgreSql 접속 비밀번호 조회 (server01 터미널에서 작업)

cat /var/lib/cloudera-scm-server-db/data/generated_password.txt

◼ 접속 (비밀번호 : vsqBur5ygT) 비민번호는 개인적으로 다 다를 수 있다.

psql -U cloudera-scm -p 7432 -h localhost -d postgres

 

◼ 테이블을 만들어 주자.

create table smartcar_symptom_info
(
  car_number varchar,
  speed_p_avg varchar,
  speed_p_symptom varchar,
  break_p_avg varchar,
  break_p_symptom varchar,
  steer_a_cnt varchar,
  steer_p_symptom varchar,
  biz_date varchar
);

 

◼ 만들어진 테이블을 확인한다.

select * from smartcar_symptom_info;

 

◼ postgreSQL jdbc driver 복사

cp /opt/cloudera/parcels/CDH/jars/postgresql-*.jar /opt/cloudera/parcels/CDH/lib/sqoop/lib