[java]embulk plugin 수정 사용 트릭

공개된 embulk jsonpath plugin 소스를 수정해서 로컬에서 사용하는 간단한 방법 정리.

• 주.
  • 특정 플러긴 수정 경험에 한정된거라 공통 사용 가능 여부는 보장할 수 없음.
  • 급박한 상황 아니라면 embulk plugin 개발 가이드 읽어 보고 도전하는게 정신 건강에 좋을 듯.

• embulk 이용한 from Rest API to DB 간단 예

1. 제목 관련 본론 들어가긴 전 사전 메모.

간단한 테스트는 해봤고 실 적용에 필요하리하 예상되는 부분을 우선 구글링해보았다.

• 파이프라인
  • in - out -> in - out -> ...
  • embulk 만으로 파이프라인 간단히 구성하는 방법은 못 찾았다.
  • airflow 를 이용한 방법이 몇 개 있던데 지금 과제에서는 배보다 배꼽이 더 커지는 셈이라 제외.
    https://qkdlql.medium.com/load-data-from-mysql-cloud-sql-to-bigquery-using-embulk-airflow-scheduling-ca215a7a5bc
• 자바 프로그램에 embulk 핸들링 하는 방법
  https://www.embulk.org/articles/2018/10/11/feedback-request-usage-of-embulkembed-and-embulkservice-binary-type.html
  • EmbulkEmbed 를 이용해서 구현하는 방법이 있는듯 한데 구체적인 레퍼런스를 못 찾았다.
  • 시간 부족으로 이 방식은 도전 과제로 남기기로 하고 스킵.
• input 데이터를 조작하는 방법
  • input 플러긴으로 입수한 데이터를 필터 플러긴을 이용해서 어느 정도 제어 가능하긴 한데 내가 필요한 필터가 아직 눈에 안 띔
  • input 플러긴을 따로 만들지 않고 스크립트를 이용해서 직접 가져오는 방법도 있던데 이건 나중에 써먹을 후보로 찜.
    https://github.com/embulk/embulk-input-script

검색에 꽤 시간 소비했는데

• 레퍼런스가 그다지 많지 않다. 게다가 그 문서들 상당수가 일본어.
• 플러긴 활용하는 방식을, 필요하면 직접 만드는걸 권장하는 분위기.
  • https://dev.embulk.org/
  • https://gist.github.com/hiroyuki-sato/b9d76bd882efa6e29ae16e01396c82ae

2. 이제 본론.

지금 당장 풀어야할 문제에 적합한 방식을 못 찾았다.
EmbulkEmbed 를 활용하는 방법이나 플러긴을 직접 작성하는건 당장은 부담스럽고.

그래서 선택한 방법이 기존 플러긴을 살짝 수정해서 사용하는 방법.

구체적인 상황으로 더 들어가보자.
지금 과제는 이전 테스트해봤던 것과 마친가지로 Rest Api 호출해서 JSON 포맷으로 결과 받아 DB에 저장하는 것.
그런데 수신한 JSON 메시지의 값을 가지고 선처리해야 할 것들이 몇가지 있다.

가령 수신한 메시지가 다음과 같다고 할 때

{"id":"OPS1","mode":"S","errCode":1, size":2, "data":[ {"block":"X1","maxRow":25,"minRow":6,...}, {"block":"X2","maxRow":12,"minRow":5,...} ] }

errCode 값을 기준으로 DB 저장 여부를 판단 처리해야 한다.

테스트에서는 증첩된 json 메시지를 파싱하기 위해 이렇게 설정했었다.

parser:
    type: jsonpath
    root: "$.data"

문제는 이 경우 data 키의 값들만 처리되어서 errCode 를 사용할 방법이 없다.

앞서 적었듯 jsonpath 플러긴에 처리 로직을 쑤셔 넣어보기로 했다.

단게별 작업 내용 요약

1. https://github.com/hiroyuki-sato/embulk-parser-jsonpath 에서 소스 다운로드.
2. 페이지에서 빌드 방법 안내한대로 'gradlew gem' 실행.
3. 윈도우 머쉰에서 빌드해서인지 'unknown encoding name - x-windows-949' 뜨길래 우선 간단히 'set GRADLE_OPTS=-Dfile.encoding=UTF-8' 명령어 실행 후 재빌드. 성공.
4. 실행 테스트.
   • % embulk run config.yml -L ./embulk-parser-jsonpath
5. 에러난다.LoadError: load error: embulk/parser/jsonpath -- java.lang.ClassNotFoundException: org.embulk.parser.jsonpath.JsonpathParserPlugin
6. 뭐냐? 막막하다 ... 일단 빌드 생성된 build 디렉토리 내부를 뒤져봤다.빌드로 생성된 embulk-parser-jsonpath-..*.jar 파일이 두 곳에 있다.
   • build\libs
   • build\gemContents\classpath
7. 각각의 상위에 embulk.bat 과 config.yml 복사해서 실행해보았다.
   • 둘 중 gemContents 에서 에러 없이 실행된다.
8. embulk-parser-jsonpath-..*.jar 파일이 있던 두 디렉토리 하위 구조를 비교해봤다.
   • build\gemContents\lib\embulk\parser\jsonpath.rb 에 다음과 같은 내용이 있다.

Embulk::JavaPlugin.register_parser(
"jsonpath", "org.embulk.parser.jsonpath.JsonpathParserPlugin",
File.expand_path('../../../../classpath', FILE))

대충봐도 classpath 를 여기서 설정해주고 있다는 것 확인되었고.

9. org.embulk.parser.jsonpath.JsonpathParserPlugin 의 run 메서드 수정(소스는 하단에).
10. gem 전체를 다시 생성할 필요는 없을테니 intellij 에서 gradle 툴 기능 이용해서 jar task 실행.
11. build\libs 에 새로 생성된 jar 파일을 build\gemContents\classpath 에 복사.
12. 다시 run 실행.
    • % embulk run config.yml -L ./gemContents
13. 잘된다.

// 본래 소스
@Override
public void run(TaskSource taskSource, Schema schema, FileInput input, PageOutput output) {
    ...
    final JsonNode json;
    try {
        json = JsonPath.using(JSON_PATH_CONFIG).parse(is).read(jsonRoot, JsonNode.class);
    }

// 수정
private Integer getErrCode(DocumentContext parseObj, String path){
    Integer errCode = -1;
    try {
        errCode = parseObj.read(path, Integer.class);
    } catch(PathNotFoundException e) {
        return errCode;
    }
    return errCode;
}

@Override
public void run(TaskSource taskSource, Schema schema, FileInput input, PageOutput output) {
    ...

    final JsonNode json;
    final Integer errCode;

    try {
        DocumentContext parseObj = JsonPath.using(JSON_PATH_CONFIG).parse(is);
        errCode = getErrCode(parseObj, "$.errCode");
        if (errCode == 1) {
            skipOrThrow(new JsonRecordValidateException(format(Locale.ENGLISH,
                "error code='%s'", errCode)), stopOnInvalidRecord);
            continue;
        }
        logger.info("#### {}", errCode);
        json = parseObj.read(jsonRoot, JsonNode.class);
    }

정리

• 특정 목적을 위해 embulk 플러긴을 작성해야할 필요가 있을 수 있다. 상세하지는 않지만 embulk 개발자 페이지에 input 플러긴 가이드 있으니 그거 참고해서 만드려는 플러긴 타입에 따라 작성하면 될거다.
• 하지만 플러그인 작성 위한 선수 지식이 없는 상태라 난 원하는 기능과 유사한 기능 제공하는 플러긴 소스 다운로드 받아서 필요한 로직만 간단히 추가해서 사용하는 방식에 도전했다.
• 간단한 기능 추가해서 내부에서만 사용할거기 때문에 gemPush 로 gem 레포지터리에 등록하거나 하지 않고 그냥 간단히 로컬 경로에 배치해서 사용할거다.

다운로드한 플러긴 소스 프로젝트 내부에서 'gradlew gem' 명령어 실행해주면
build\gemContents\ 하위에 classpath와 lib 디렉토리 생성되고 여기에 실행에 필요한 파일들이 위치하게된다.

꼭 실행에 필요한 파일들은 다음과 같다.

• classpath 하위에 생성된 참조 라이브러리들과 해당 플러긴 기능이 구현된 jar 파일.
• lib\embulk\parser 에 생성된 루비 스크립트 파일(플러긴 타입에 따라 input?, output?, parsert, filter? 등의 이름으로 디렉토리 생성되지 않을까 추측된다)

다운로드한 소스 수정한 뒤 jar 생성해서 (내 경우 build\libs 에 생성됨) 위 classpath 디렉토리에 복사해 넣는다.

classpath, lib 두 디렉토리만 실행 원하는 곳에 복사한 뒤 그 상위 디렉토리에서 다음과 같은 식으로 실행한다.
embulk run config.yml -L ./디렉토리명

실행 로그에 [INFO] (0001:transaction): Loaded plugin embulk/... 식으로 플러긴명이 아니라 경로가 표시된다면 일단 본인이 작업한 로컬 플러긴이 실행되고 있음을 확인할 수 있다.

참고. gem 인스톨했던 플러긴과 소스 수정한 로컬 플러긴을 이용할 때의 런타임 로그를 비교.

// gem 인스톨했던 플러긴 사용 시
[INFO] (main): Started Embulk v0.9.23
[INFO] (0001:transaction): Loaded plugin embulk-input-http (0.24.0)
[INFO] (0001:transaction): Loaded plugin embulk-parser-jsonpath (0.3.2)

//소스 수정한 로컬 플러긴 사용 시
[INFO] (main): Started Embulk v0.9.23
[INFO] (0001:transaction): Loaded plugin embulk-input-http (0.24.0)
[INFO] (0001:transaction): Loaded plugin embulk/parser/jsonpath from a load path