좌충우돌 개발 블로그

Logstash 부하로 인하여 Elastic Search에 업데이트가 안되고 있었다.

그래서 부랴부랴 특정 토픽들을 제외하여 급한 로그들을 처리했다.

그런데 일부 로그들을 파싱하는 과정에서 시간 정보를 현재 시간으로 쓰는 경우가 있었다.

시간 정보를 따로 갖고는 있었는데 해당 필드로 처리하고 있지 않고 있었다.

이것을 해결하기 위한 방법을 찾아보니 _update_by_query 를 사용하면 된다고 하여 시도해보았다.

처음엔 인덱스 전부를 업데이트해보았으나 너무 오래 걸렸다.

그래서 특정 시간대의 로그만 처리하기로 했고 다음과 같이 데이터를 보내 해결할 수 있었다.

{
  "script": {
    "lang": "painless",
    "source": "ctx._source['@timestamp'] = ctx._source.time"
  },
  "query": {
    "bool": {
        "must": [
            {
                "exists": {
                    "field": "time"
                }
            },
            {
                "range": {
                    "@timestamp": {
                        "time_zone": "+09:00",        
                        "gte": "2020-11-20T21:25:00", 
                        "lte": "2020-11-20T21:30:00"                  
                    }
                }
            }
        ]
    }
  }
}

참고 문헌

1. https://stackoverflow.com/questions/55570214/how-do-i-overwrite-the-timestamp-field-with-another-field-in-elasticsearch

2. https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html

3. https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-range-query.html

4. https://www.elastic.co/guide/en/elasticsearch/reference/current/compound-queries.html

ElasticSearch의 인덱스를 불러와서 Zeppelin에서 사용하려고 했다.

인덱스를 제목-날짜(yyyy.MM.dd) 형식으로 만들어서 날짜를 항상 오늘 날짜로 사용하고 싶었다.

이를 위해서 현재 시간 정보를 불러오는 함수가 필요했는데 다음과 같이 사용했다.

val ts = (current_timestamp()).expr.eval().toString.toLong
val dateValue = new java.sql.Timestamp(ts/1000).toLocalDateTime.format(java.time.format.DateTimeFormatter.ofPattern("yyyy.MM.dd"))

어제 정보도 불러오면 좋을 것 같아 다음과 같이 하여 쉽게 불러올 수 있었다.

val ts = (current_timestamp()).expr.eval().toString.toLong - 86400000L
val dateValue = new java.sql.Timestamp(ts/1000).toLocalDateTime.format(java.time.format.DateTimeFormatter.ofPattern("yyyy.MM.dd"))

특정 데이터를 기준으로 정렬을 하거나 조건으로 조회하려고 했는데 해당 데이터가 문자열이었다.

그래서 정렬을 제대로 하기 위해 형변환이 필요했는데 실수형으로 변환해야 했다.

withColumn과 cast로 쉽게 해당 열의 타입을 변환할 수 있었다.

import org.apache.spark.sql.types.FloatType

...
df.withColumn("etc", df("etc") cast FloatType).createOrReplaceTempView("temp")

참고 문헌

1. https://stackoverflow.com/questions/55267527/spark-getting-current-date-in-string/55267586

2. https://stackoverflow.com/questions/52590524/sparkstream-convert-string-to-float-in-sql-query

이번에 프로젝트를 수행하면서 일별로 데이터를 분석할 필요가 있었다.

처음엔 수동으로 작업을 했는데 자동으로 구동되면 좋을 것 같아서 찾아보니 역시나 기능이 있었다.

$ZEPPELIN_HOME/conf/zeppelin-site.xml 에서 주석으로 돼있는 내용을 풀고 다음과 같이 수정하면 된다.

(zeppelin.notebook.cron.folders 에도 값이 있어야 하는데 / 나 특정 디렉토리를 입력했을 때는 버튼이 나오지 않았다.)

<property>
<name>zeppelin.notebook.cron.enable</name>
<value>true</value>
<description>Notebook enable cron scheduler feature</description>
</property>
<property>
<name>zeppelin.notebook.cron.folders</name>
<value>*</value>
<description>Notebook cron folders</description>
</property>

이미 설정된 값은 None, 1m, 5m, 1h, 3h, 6h, 12h, 1d로 돼있다.

그외에 설정을 하려면 크론 설정은 Quatz에서 하는 것과 같이 하면 된다.

10분마다로 설정을 하려며 0 0/10 * * * ? 로 하면 된다.

참고 문헌

1. https://zeppelin.apache.org/docs/0.8.0/usage/other_features/cron_scheduler.html

2. https://stackoverflow.com/questions/58727773/cant-turn-on-cron-feature-in-apache-zeppelin

3. https://www.quartz-scheduler.org/documentation/quartz-2.3.0/tutorials/crontrigger.html#examples

ElasticSearch에서 다음과 같은 로그가 발생했었다.

[WARN ][o.e.d.i.m.MapperService  ] [_default_] mapping is deprecated since it is not useful anymore now that indexes cannot have more than one type

이 로그는 _default_ mapping을 사용해서 발생하는 문제로

6.0.0 부터 Deprecated 되어 실제 적용된 매핑 타입으로 적용하면 된다.

그러나 "실제 적용된 매핑 타입으로 적용"을 어떻게 해야하는지 찾을 수 없었다.

mapping api로 되는지도 해봤지만 전혀 되지 않았고 Kibana 설정으로도 변경을 할 수 없었다.

인덱스를 새로 만들어서 복사해보라는 것도 있어 이를 해봤지만 이것 역시 _default_가 생겼다.

그러다가 다른 인덱스에서는 발생을 안하는데 logstash 로 시작하는 인덱스에서만 _default_가 있어

인덱스 이름에 logstash 를 다른 이름으로 변경하여 저장해보니 _default_가 만들어지지 않았다.

정확한 이유는 모르겠지만 logstash로 시작되는 인덱스의 경우에는 _default_가 생성되는 것으로 보여진다.

참고 문헌

1. https://www.elastic.co/guide/en/elasticsearch/reference/6.4/default-mapping.html