Elasticsearch와 Apache Hadoop

Hadoop vs Logstash/Elasticsearch

• Hadoop
  • 구글과 같이 아주 많은 웹사이트에서 데이터를 수집해야할 때 적절
  • 효율적으로 빅데이터를 수집 및 처리할 수 있다.
• Elasticsearch
  • 추후에 빠르고 효율적인 검색을 위해 데이터를 저장하고 인덱스화 해야하는 경우 적절
  • 뛰어난 검색 기능 분석과 데이터 시각화에 적합
• Logstash - 실시간 데이터를 수집하고자 할 때 적절
  각 프레임워크을 함께 사용하면 장점을 같이 가져갈 수 있다. 예를 들어 수십억개의 웹사이트에 대한 정보를 검색하는 예시에서 하둡을 사용하여 데이터를 수집하고 이를 Elasticsearch에 전달한다. 그러면 Elasticsearch는 해당 데이터를 검색하는 사용자에게 신속하게 결과를 반환할 수 있다.

Hadoop 설치

먼저 실습하고 있는 가상환경에 hadoop을 설치하도록 한다. elasticsearch가 설치되었기 때문에 필요한 jdk는 설치되었다고 가정한다.

• hadoop과 관련된 작업을 진행할 유저를 별도로 생성해준다.
  • sudo adduser hadoop
    • 이 과정에서 password는 별도로 기억해두도록 하자. 하둡과 관련된 작업을 실시할 때는 하둡 유저로 로그인을 해야하기 때문이다. password 외에 다른 설정은 그냥 엔터를 눌러서 진행한다.
  • sudo usermod -aG sudo hadoop
    • 생성한 hadoop 이라는 유저에 sudo 권한을 부여한다.
  • su - hadoop
    • 잘 생성되었는지 로그인해본다.
    • 아래의 설치작업을 해당 계정에서 실시하도록 한다.
• apache용 hadoop을 다운로드받는다.
  • wget https://downloads.apache.org/hadoop/common/hadoop-3.2.4/hadoop-3.2.4.tar.gz
  • 사용 버전은 apache hadoop download 페이지에서 버전을 확인하거나 특정 버전을 위의 명령어의 버전 부분에 맞게 수정해서 입력해주면 된다.
• 다운로드가 완료되면 압축을 풀고 경로를 이동시킨다.
  • tar -xvzf hadoop-3.2.4.tar.gz
  • sudo mv hadoop-3.2.4 /usr/local/hadoop
  • hadoop 사용자 및 그룹이 /usr/local/hadoop 디렉토리를 이미 소유하고 있어야 한다. 실제로 이 사용자와 그룹의 소유인지 확인하기 위해 다음 명령을 실행한다.
  • sudo chown -R hadoop:hadoop /usr/local/hadoop
• 환경 변수 설정
  • hadoop은 사용하는 디렉토리 경로를 파악하기 위해 환경변수를 사용한다. 이를 설정해준다.
  • vim ~/.bashrc
  • 파일의 마지막으로 이동한 후 다음을 입력한다.
  • export HADOOP_HOME=/usr/local/hadoop
  • export HADOOP_INSTALL=$HADOOP_HOME
  • export HADOOP_MAPRED_HOME=$HADOOP_HOME
  • export HADOOP_COMMON_HOME=$HADOOP_HOME
  • export HADOOP_HDFS_HOME=$HADOOP_HOME
  • export YARN_HOME=$HADOOP_HOME
  • export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
  • export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin
  • export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib/native"
  • 저장 후 환경변수를 적용하기 위해 source ~/.bashrc 를 입력해주자.
• Hadoop Config 설정
  • 하둡은 필요한 java 구성 요소를 찾을 수 있는 곳을 알아야 한다. 따라서 JAVA_HOME 환경변수를 사용하여 올바른 위치를 입력한다.
  • readlink -f $(which javac)
    • 해당 명령어에서 에러가 발생한다면 전체 JDK가 설치되지 않은 것이니 설치가 필요하다.
    • sudo apt install default-jdk
    • 설치 이후 다시 시도해보면 경로를 반환한다.
  • 반환한 경로 중 /usr/lib/jvm/java-11-openjdk-amd64 까지의 부분을 복사한다.
  • sudo vim $HADOOP_HOME/etc/hadoop/hadoop-env.sh 파일을 열고 'JAVA_HOME' 을 찾아서 주석을 해제하고 복사한 경로를 입력해주고 저장하고 나간다.
  • 이후 hadoop version 을 입력했을 때 설치한 하둡 버전이 명시되면서 부가적인 정보가 출력되면 성공이다.

MapReduce 프로젝트 생성

Apache 결합 로그 형식으로 생성된 Index 혹은 Sample Access Log 파일은 Maven 빌드 도구를 사용하여 MapReduce 코드를 jar 파일로 컴파일한다. 실제로는 프로젝트를 생성하고 필요한 코드를 작성하려면 IDE를 설치해야한다. 그런 다음 로컬환경에서 Maven으로 프로젝트를 컴파일하고 jar 파일로 로컬 환경에서 하둡 인스턴스로 전송한다.
elasticsearch와 hadoop을 연동하기 위한 프로젝트를 생성해보자.
디렉토리 구조는 다음과 같다.

각 파일은 다음과 같이 구성되어 있다.

• pom.xml

<?xml version="1.0" encoding="UTF-8"?>  
<project xmlns="http://maven.apache.org/POM/4.0.0"  
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
    <modelVersion>4.0.0</modelVersion>  
  
    <developers>  
        <developer>  
            <name>Oliver Mascarenhas</name>  
        </developer>  
    </developers>  
  
    <groupId>com.coralogix</groupId>  
    <artifactId>eswithmr</artifactId>  
    <version>1.0-SNAPSHOT</version>  
  
    <properties>  
        <maven.compiler.source>1.8</maven.compiler.source>  
        <maven.compiler.target>1.8</maven.compiler.target>  
    </properties>  
  
  
    <dependencies>  
        <dependency>  
            <groupId>org.apache.hadoop</groupId>  
            <artifactId>hadoop-client</artifactId>  
            <version>3.2.1</version>  
        </dependency>  
        <dependency>  
            <groupId>org.elasticsearch</groupId>  
            <artifactId>elasticsearch-hadoop-mr</artifactId>  
            <version>7.8.0</version>  
        </dependency>  
        <dependency>  
            <groupId>commons-httpclient</groupId>  
            <artifactId>commons-httpclient</artifactId>  
            <version>3.1</version>  
        </dependency>  
    </dependencies>  
  
  
    <build>  
        <plugins>  
            <plugin>  
                <groupId>org.apache.maven.plugins</groupId>  
                <artifactId>maven-shade-plugin</artifactId>  
                <version>3.2.4</version>  
                <executions>  
                    <execution>  
                        <phase>package</phase>  
                        <goals>  
                            <goal>shade</goal>  
                        </goals>  
                        <configuration>  
                            <transformers>  
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">  
                                    <manifestEntries>  
                                        <Main-Class>com.coralogix.AccessLogIndexIngestion</Main-Class>  
                                        <Build-Number>123</Build-Number>  
                                    </manifestEntries>  
                                </transformer>  
                            </transformers>  
                        </configuration>  
                    </execution>  
                </executions>  
            </plugin>  
        </plugins>  
    </build>  
  
  
</project>

• AccessLogIndexIngestion.java

package com.coralogix;  
  
  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.NullWritable;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.mapreduce.Mapper;  
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
import org.elasticsearch.hadoop.mr.EsOutputFormat;  
import org.elasticsearch.hadoop.util.WritableUtils;  
  
import java.io.IOException;  
import java.util.LinkedHashMap;  
import java.util.Map;  
  
public class AccessLogIndexIngestion {  
  
    public static class AccessLogMapper extends Mapper {  
        @Override  
        protected void map(Object key, Object value, Context context) throws IOException, InterruptedException {  
  
            String logEntry = value.toString();  
            // Split on space  
            String[] parts = logEntry.split(" ");  
            Map<String, String> entry = new LinkedHashMap<>();  
  
            // Combined LogFormat "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-agent}i\"" combined  
            entry.put("ip", parts[0]);  
            // Cleanup dateTime String  
            entry.put("dateTime", parts[3].replace("[", ""));  
            // Cleanup extra quote from HTTP Status  
            entry.put("httpStatus", parts[5].replace("\"",  ""));  
            entry.put("url", parts[6]);  
            entry.put("responseCode", parts[8]);  
            // Set size to 0 if not present  
            entry.put("size", parts[9].replace("-", "0"));  
  
            context.write(NullWritable.get(), WritableUtils.toWritable(entry));  
        }  
    }  
  
  
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {  
        Configuration conf = new Configuration();  
        conf.setBoolean("mapred.map.tasks.speculative.execution", false);  
        conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);  
        conf.set("es.nodes", "127.0.0.1:9200");  
        conf.set("es.resource", "logs");  
  
        Job job = Job.getInstance(conf);  
        job.setInputFormatClass(TextInputFormat.class);  
        job.setOutputFormatClass(EsOutputFormat.class);  
        job.setMapperClass(AccessLogMapper.class);  
        job.setNumReduceTasks(0);  
  
        FileInputFormat.addInputPath(job, new Path(args[0]));  
        System.exit(job.waitForCompletion(true) ? 0 : 1);  
    }  
  
}

이제 이 프로젝트를 mvn clean package 명령어를 사용해 build해준다.

Elasticsearch Index Ingestion

이제 실제로 데이터를 하둡을 통해서 넣어보자. 먼저 실습에 필요한 예제를 다운로드 받는다.

• wget https://raw.githubusercontent.com/linuxacademy/content-elastic-log-samples/master/access.log
• 컴파일 후 생긴 jar 파일을 다운롤드 받은 로그 파일과 동일한 위치로 옮겨준다.
  • cp target/eswithmr-1.0-SNAPSHOT.jar .
• 파일을 실행시켜준다.
  • hadoop jar eswithmr-1.0-SNAPSHOT.jar access.log 
    • 이렇게 표시되면 된다. Map input, output records 수가 10000개임을 확인할 수 있다.

• 잘 들어갔는지 확인한다.
  • curl localhost:9200/_cat/indicies?v
  • log 라는 이름의 파일이 10000개로 표시된다면 정상적으로 Index화 된 것이다.