개발자

스프링 프로젝트 생성 시 applicationContext.xml에서 beans 에러가 발생하는 경우

오류 : 

beans xmlns="http://www.springframework.org/schema/beans"

--> URI is not registered (Settings | Languages & Frameworks | Schemas and DTDs)

해결 방법 : 해당 위치에서 alt + enter > Fetch external resource 선택

아주 잘 설명되 있는 스토리가 있어서 공유한다.

https://whitepaek.tistory.com/55

https://whitepaek.tistory.com/56

빅데이터 : Volume. Veriety, Velocity -크기, 다양성, 속도

Variability(가벼성), Veracity(정확성), Complexity(복잡성), Visibillity(시인성)

스파크 : 하둡 기반의 맵리듀스 작업의 단점(파일방식,속도느림,복잡성)을 보완하기 위해 개발, 메모리 방식

데이터 모델 1. RDD (Resilient distributed dataset) : 분산 방식으로 저장된 데이터 요소들의 집합, 병렬처리가 가능하고 장애 복구 내성(tolerance)를 가진 모델

• RDD는 파티션 단위로 나뉘어 병렬 처리 됨.
• 작업중 파티션은 재구성 되면서 네트워크를 통해 다른 서버의 파티션으로 이동하는 셔플링이 발생. 셔플링은 작업 성능에 큰 영향을 주기 때문에 주의해야 함.
• RDD는 연산이 수행 시 새롭게 생성되며, 변경되지 않는 읽기 전용 모델이다. 전 단계 RDD와 어떤 연산을 적용해서 만들어 지는지를 기억(lineage)하고 있어서 특정 절차에 장애가 발생해도 해당 이력을 바탕으로 RDD 생성을 재시도 하게 된다.
• RDD 제공 연산 1. - 트랜스포메이션(Transformation) : 새로운 RDD를 만들어 내는 변환 연산, 액션 연산이 호출될 때 여러 변환 연산들의 최적화를 분석해서 실행(lazy 실행방식)이 된다.
• RDD 제공 연산 2. - 액션(Action) : 연산의 결과(RDD가 아닌 다른 값, 반환하지 않는 연산 포함)를 반환하는 동작, ex) sum(), stddev() 등

DAG 스케쥴러 : 셔플의 유무에 따라 스테이지를 나누고 셔플로 인해 발생하는 부하를 최대한 줄이는 역활을 함.

• Driver 메인 함수 > Spark Context > RDD 연산 정보를 DAG 스케쥴러에게 전달 : 전체 데이터 처리 흐름을 분석해서 네트워크를 통한 데이터 이동이 최소화 되도록 Stage 구성 > 클러스터 매니저에게 전달 > Stage 별 task 실행
• 좁은 의존성 : 연산이 실행되는데 필요한 파티션이 적은 경우, 파티션간의 셔플이 발생하지 않음. (더하기 연산)
• 넓은 의존성 : 여러 파티션의 정보가 연산에 필요한 경우, 각 파티션들의 데이터를 셔플하여 시간이 오래 걸림. (SUM 등)

Spark Context와 Spark Session : 스파크 애플리케이션이 동작하기 위한 서버 프로세스가 필요. 

• 클러스터를 구성하는 개별 서버 간에 메시지와 데이터를 주고 받을 수 있는 채널(백엔드 서버 프로세스) 구동
• 개별 작업 수행에 필요한 메타 정보를 저장하고 관리
• 스파크 2.0부터 세션안에 컨텍스트가 포함됨.

데이터 프레임

• RDD는 실행할 함수를 직접 전달 받아 사용하는 대신에 DataFrame은 사전에 정의된 표현식과 내부 함수를 이용해 코드를 작성
• 논리적 작업 트리로 변환한 후 최적화 과정을 거쳐서 최종 동작을 수행 => RDD에 비해 훨씬 효율적이고 최적화를 수행.

RDD : 직접 함수를 전달받아 처리
val words = rdd.flatMap(str => str.split(" "))
val wcPair = words.map(v => (v,1))

DF : 내장 함수 사용
val wordDF = df.select(explode(split(col("value"), " ")).as("word"))

input {
	aggregation {
    	hosts => "host:port"
        index => "keyword_index"
        type => "keyword_qc"
        schedule => "10 4 * * *"
        aggs_name => "keyword_cnt"
        docinfo => true
        query_filepath => "/etc/logstash/keyword_cnt.es"
        parameters => [
        	{ "_day_" => "1" "_size_" => "1234" "_bool_" => "true" "_astric_" => "*" },
        	{ "_day_" => "2" "_size_" => "2345" "_bool_" => "false" "_astric_" => "*" }
        ]
        result_has_parameters => true
    }
    
    jdbc {
    	connection_retry_attempts => 3
        connection_retry_attempts_wait_time => 1
        jdbc_driver_library => "/etc/logstash/lib/mysql-connector-java-version.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_validate_connection => false
        jdbc_connection_string => "jdbc:mysql://host:port/db"
        jdbc_user => "simon"
        jdbc_password => "simon_work1004"
        jdbc_fetch_size => 1000
        statement_filepath => "/etc/logstash/keyword_rank.sql"
        type => "keyword_buy"
        schedule => "30 2 * * *"
        last_run_metadata_path=>"/etc/logstash/.logstash_jdbc_last_run"

    }
}

filter {
	if [type] == "keyword" {
    	 ruby {
            code => "
                from = event.get('parameters')['_day_'].to_i
                dateTime = (DateTime.new(Time.now.year, Time.now.month, Time.now.day, 0, 0, 0, 0) - from)
                date = dateTime.strftime('%s')
                event.set('date_index', dateTime.strftime('%y%m'))
                id = date + '_' + [event.get('key')].pack('m0')
                event.set('searchKeyword', event.get('key'))
                event.set('keyword', event.get('key').strip.upcase)
                event.set('date', date)
                event.set('id', id)

                platform = event.get('parameters')['_platform_']

                if platform == '*'
                    platform = 'total'
                end

                obj = Hash.new
                obj['keyword1_cnt'] = event.get('keyword2_cnt')

                event.set(platform, obj)
            "
        }

        mutate {
            replace => {
                "[@metadata][_index]" => "keyword_%{date_index}"
                "[@metadata][_type]" => "keyword"
                "[@metadata][_id]" => "%{[id]}"
                "[tags]" => "%{[type]}"
            }
            remove_field => [ "@timestamp", "@version", "type", "doc_count", "id", "parameters", "date_index", "key" ]
        }

        prune {
            whitelist_names => [ 'web', 'app', 'total', 'date', 'keyword', 'tags' ]
        }

    }
    else if [type] == "keyword2" {
    
    }
}

output {
	if "keyword" in [tags] or "keyword2" in [tags] {
    	elasticsearch {
        	action => "update"
            hosts => 
            index => "%{[@metadata][_index]}"
            document_type => "~"
            document_id => "~"
            doc_as_upsert => true
            template_name => "keyword_temp"
            template => "/etc/logstash/template/keyword_temp.json"
            template_overwrite => true
            retry_on_conflict => 5
        }
    }
}