package com; import java.sql.*; import java.util.HashMap; public class App { private static Connection connect; private static Statement statement; private static PreparedStatement preparedStatement; private static ResultSet resultSet; private static String serverURL; static { connect = null; statement = null; preparedStatement = null; resultSet = null; serverURL = "jdbc:phoenix:12.234.56.78:218..

기존 HBase 테이블에 Phoenix 테이블을 매핑하는 법은 기존에 있는 HBase 테이블과 똑같은 이름의 Phoenix테이블 또는 뷰를 생성한다. 또한 두 테이블 컬럼의 자료형이 일치해야하는데, 특히 숫자의 경우 Hbase와 Phoenix의 표현방식이 다르므로 표현되기 어렵다. "Phoenix will flip the first bit so that negative values will sort before positive values. Because HBase sorts row keys in lexicographical order and negative value’s first bit is 1 while positive 0 so that negative value is ‘greater than’ p..

[ 아키텍처 ]Kafka -> Nifi -> HBase -> Python으로 데이터 Fetch   [ Kafka ]카프카의 토픽명과 컨슈머그룹명을 정해 데이터를 받는다.  [ Nifi ]아래와 같은 프로세스 플로우로 구성할 수 있다. ConsumerKafka에 토픽명과 그룹명을 설정값으로 입력한다UpdateAttribue와 ReplaceText는 해당 데이터가 나이파이에 적제되는 시간을 입력하기 위함이므로 생략가능하다.PutHBaseJson에 HBase TableName, Row Identifier, Column Family값을 입력한다.네임스페이스가 있다면 NS:TableName과 값이 입력한다.Row Identifier값은 로우키가 되므로 유니크한 값으로 설정한다. 예시에선 $(createdAt}_$..

1. 날짜 포맷 변환from_unixtime(bigint unixtime, string format)return 값 : stringunix타임스탬프 값을 정해진 날짜형식에 맞추어 출력한다.- select from_unixtime(1323308938, 'yyyy-MM-dd') from dual;20210101포맷을 2021-01-01포맷으로 날짜형식변환도 가능하다.- select from_unixtime(unix_timestamp('20210101, 'yyyyMMdd'), 'yyyy-MM-dd') from dual; 2. 특정 날짜 값 가져오기날짜 부분만 리턴select to_date('2021-01-13 10:11:34') from dual;-> 2021-01-13연도 부분만 리턴select year('..

Sqoop의 Default 구분자는 ',' 이며, RDB 데이터를 입력받을 시 Column의 Text데이터가 "안녕하세요, 저는 어쩌고, ..." 와 같은 ','가 포함되었다면 구분자를 변경해야 한다. 아래 설정은 \t (탭)으로 구분자를 설정하며 Oozie XML -> Sqoop -> Hive Table에 넣는 과정을 담는다. [ Oozie XML ] 아래 설정을 추가 한다. --fields-terminated-by "\t" [ Hive Table Create ] 생성시 Delimited fields를 정의한다. ( 또는 업데이트 ) CREATE TABLE temp.table_temp ( id BIGINT, name VARCHAR(30), pay DECIMAL(10,2) tag VARCHAR(200) ..

'20528978에 대하여 인코딩 디코딩. [ 인코딩 ] select base64(encode('20528978', 'UTF-8')) 결과 값 : MjA1Mjg5Nzg= [ 디코딩 ] select decode(unbase64('MjA1Mjg5Nzg='), 'UTF-8') 결과 값 : 20528978 [ 다양한 형식 ] select decode(unbase64('MjA1Mjg5Nzg='), 'ISO-8859-1'); select decode(unbase64('MjA1Mjg5Nzg='), 'UTF-16LE'); 변환해주는 웹사이트 https://www.base64encode.org/d

Kafka에서 데이터를 받아 HDFS에 json, csv형태로 입력받는다.저장 후 Hive, Spark 등으로 데이터를 처리할 수 있으며실시간 데이터 인풋을 Nifi를 통해 확인할 수 있는 장점이 있다.  1. [ ConsumeKafka ] 컨슈머로 연결하여 브로커에서 보내는 메세지를 소비Properties : Kafka Brokers : 172.31.11.11:9092, 172.31.22.22:9092, 172.31.33.33:9092Topic Name(s) : your-topic-nameGroup ID : your-kafka-group-id  2. [ UpdateAttribute ]들어오는 카프카데이터에 createdAt를 추가하기 위한 프로퍼티filename은 HDFS저장시 파일명을 시간으로 지정하..

[ Ubuntu ] * 사전설치 Python3 sudo apt-get upgrade # Pip설치 sudo apt-get install python3-pip #Kafka Lib 설치 sudo pip3 install kafka-python [ 테스트 코드 ] from kafka import KafkaProducer import time producer = KafkaProducer( bootstrap_servers=['172.31.11.11:9092', '172.31.22.222:9092', '172.31.33.33:9092'] ) for i in range(1,50): producer.send('Topic-Name', str.encode('kafka:-%d' % i) ) time.sleep(0.5)