[Elastic] 지식 그래프란?

Relation DataBase(테이블 형식의 데이터 베이스)와 다르게 지식을 그래프 형식으로 표현하여 새로운 정보 추론과 여러 가지 속성을 확장할 수 있는 새로운 종류의 그래프 데이터 베이스이다.

• 검색 품질을 향상 시킬 수 있다. -> 검색을 확장할 수 있다.
• 아이디어, 컨셉들의 관계를 그래프로 나타낸다.
• Entity들 사이에서 어떤 관계가 있는지 파악할 수 있다.
• 화살표로 연관 관계를 표현해 준다. -> 여러 가지 Entity에 형성되는 메타 데이터를 나타낼 수 있다.

이런 지식 그래프 없이 Relational Database로 Entity들의 관계를 표현하려면 무수히 많은 테이블이 필요하다.

** 지식 그래프는 Entity들을 하나씩 정리해 주고, Entity들 사이에서의 관계를 표현해 줌으로써 매우 효율적인 데이터 저장소 역할을 할 수 있다.

예를 들어, 검색 엔진에 '어느 대륙에 숭례문이 위치해 있나요?'라는 문장을 입력하면 숭례문 -> 대한민국 -> 아시아 -> 대륙 확인 후, 아시아라는 검색 결과가 빠르게 가능하다.

즉, 검색 문장에 존재하지 않는 정보들을 관계를 따라가면서 검색 결과를 가져올 수 있다.

---

* 지식 그래프 데이터 베이스 종류

• Neo4j
• RedisGraph

Relational Database에 저장이 가능하지만 더 빠르게 연관 관계를 따라갈 수 있게 해주는 전문적인 그래프 데이터 베이스를 사용한다.

 

* 지식 그래프 사용 사례

• 관계를 표현하여 가장 관련된 정보들을 먼저 보여주고자 할 때
• 하나의 카테고리에 연관된 정보들을 보여주고자 할 때
• 여러 가지의 요소들이 합쳐져 어떤 정보를 나타낼 때
• 한 가지 정보를 토대로 연관된 모든 정보를 보여주고자 할 때

** 정보를 검색하고 결과를 반환할 때, 연관 관계를 고려하여 추론을 하고 추론 결과를 바탕으로 사용자에게 더 적합한 정보를 응답할 수 있다. -> 연관 관계 사이의 점수 중 낮은 점수는 Relevancy가 낮고, 높은 점수는 Relevancy가 높다는 것을 활용한다.

 

키워드 확장 예제

** 이 방법은 지식 그래프를 사용한 사례는 아니지만 지식 그래프의 개념을 확인해 볼 수 있는 예제이다.

1. 인덱스 생성

 

| 코드

import requests
import json

url = "http://localhost:9200/products"

payload = json.dumps({
    "settings": {
        "index": {
            "number_of_shards": 1,
            "number_of_replicas": 1
        },
        "analysis": {
            "analyzer": {
                "analyzer-name": {
                    "type": "custom",
                    "tokenizer": "keyword",
                    "filter": "lowercase"
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "id": {
                "type": "long"
            },
            "content": {
                "type": "text"
            },
            "title": {
                "type": "text"
            },
            "url": {
                "type": "text"
            },
            "image_file": {
                "type": "text"
            },
            "post_date": {
                "type": "date"
            },
            "modified_date": {
                "type": "date"
            },
            "shipped_from": {
                "type": "text"
            },
            "keywords": {
                "type": "text"
            },
            "meta_data": {
                "type": "object"
            }
        }
    }
})
headers = {
    'Content-Type': 'application/json'
}

response = requests.request("PUT", url, headers=headers, data=payload)

print(response.text)

다음 코드를 사용하여 새로운 Index를 생성한다.

 

2. 데이터 넣기

 

| 코드

* 위키미디아 파일에서 데이터 처리(https://dumps.wikimedia.org/kowiki/)

import mwparserfromhell
import re
import xml.etree.ElementTree as etree

def loadWikimedia(source_file):
    tree = etree.parse(source_file)
    root = tree.getroot()
    namespace = getNamspace(root.tag)
    kg = {}
    for page in root.findall('./' + namespace + 'page'):
        title = page.find(namespace + 'title').text
        page_content = page.findall(
            './' + namespace + 'revision/' + namespace + 'text')
        entry = {}
        if len(page_content) > 0:
            wikicode = mwparserfromhell.parse(page_content[0].text)
            templates = wikicode.filter_templates()
            for template in templates:
                #print(template.name)
                for param in template.params:
                    value = stripWikilinksForText(str(param.value)).strip()
                    if len(value) > 0:
                        entry[str(param.name).strip()] = value
        kg[title] = entry
    return kg

# 위키미디아 스타일 링크에서 텍스트만을 추출하는 helper function
def stripWikilinksForText(wikilink):
    return re.sub(r'\[\[(.+?)\|.+?\]\]', r'\1', wikilink).replace('[[', '').replace(']]', '')

# XML 의 namespace 를 찾아 돌려준다.
def getNamspace(tag):
    return '{' + tag.split('}')[0].strip('{') + '}'

kg 변수에는 1차적인 맵들만 존재한다.

import ast
import datetime
import hashlib
import json
import mysql.connector
import requests
import kg.kg_loader as kg_loader

# SQL 에 연결하여 제품 페이지들을 추출하여 ProductPost array 로 돌려주는 함수
def getPostings():
    kg_source = 'kg/kowiki-20210701-pages-articles-multistream-extracted.xml'
    wiki_kg = kg_loader.loadWikimedia(kg_source)
    cnx = mysql.connector.connect(user='root',
                                password='비밀번호',
                                host='localhost',
                                port=9906,
                                database='데이터베이스')
    cursor = cnx.cursor()

    query = ('쿼리')
    cursor.execute(query)

    posting_list = []
    for () in cursor:
        print("Post {} found. URL: {}".format(id, url))
        meta_data = {}
        keywords = []
        # 1. 각 단어마다 위키미디아에 관련 정보를 찾는다.
        for n_gram in title.split():
           if n_gram in wiki_kg:
               print("found entry for " + n_gram)
               meta_data = {**meta_data, **wiki_kg[n_gram]}
               subspecies = maybeGetSubspecies(wiki_kg[n_gram])
               if subspecies != None:
                   keywords.append(subspecies)
        product = ProductPost(id, content, title, url,
                              post_date, modified_date, assumeShippingLocation(meta_value), image, meta_data, " ".join(keywords))
        posting_list.append(product)

    cursor.close()
    cnx.close()
    return posting_list

# 2. 위키미디아 데이타에 특정한 attribute을 추출
# Subspecies field 가 해당 wiki data 에 존재하면 돌려주고, 아니면 None 을 돌려준다.
def maybeGetSubspecies(wiki_page):
    sub_species_key = u'과'
    if sub_species_key in wiki_page:
        return wiki_page[sub_species_key]
    return None

# 엘라스틱서치에 출력하는 함수
def postToElasticSearch(products):
    putUrlPrefix = 'http://localhost:9200/products/_doc/'
    headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
    for product in products:
        id = str(product.id)
        print(id)
        r = requests.put(putUrlPrefix + id, data=json.dumps(product.__dict__,
                         indent=4, sort_keys=True, default=json_field_handler), headers=headers)
        if r.status_code >= 400:
            print("There is an error writing to elasticsearch")
            print(r.status_code)
            print(r.json())

# datetime -> iso 형식으로
def json_field_handler(x):
    if isinstance(x, datetime.datetime):
        return x.isoformat()
    raise TypeError("Unable to parse json field")

# 제품 페이지를 표헌하는 class
class ProductPost(object):
  def __init__(self, id, content, title, url, post_date, modified_date, shipped_from, image_file, meta_data, keywords):
    self.id = id
    self.content = content
    self.title = title
    self.url = url
    self.post_date = post_date
    self.modified_date = modified_date
    self.shipped_from = shipped_from
    self.image_file = image_file
    self.meta_data = meta_data
    self.keywords = keywords

p = getPostings()
postToElasticSearch(p)

• mwparserfromhell -> 위키미디아의 데이터를 쉽게 parser 해주는 하나의 모듈
• DataBase에 저장된 정보에 위키미디아에서 가져온 정보의 '과' 항목을 keyword 필드에 저장한다.

해당 파일을 실행해서 Elastic Search에 데이터를 저장한다.

http://localhost:9200/products/_search

Elastic Search에 저장된 데이터를 확인했을 때, 다음과 같은 결과를 확인할 수 있다.

keyword에 국화과가 저장된 것을 확인할 수 있다.

 

3. 키워드 검색

http://localhost:9200/products/_search?q=국화과 

* 국화과라는 키워드로 검색할 경우 id 14번이 검색 결과로 도출되었다. -> 키워드가 확장된 것을 확인할 수 있다.

 

** 키워드를 추가하였을 뿐, 효율적인 그래프 탐색을 할 수는 없다.

* 실전에서는 문서 자체에 키워드를 확장하지 않는다. -> Neo4j나 RedisGraph를 사용해서 키워드 검색 쿼리가 왔을 때, 키워드 관련 속성을 확장해서 검색 결과를 제공한다.

* 지식 그래프를 통해서 검색 결과를 향상시킬 수 있고, 개인화, 여러 가지 새로운 추론을 나타냄으로써 새로운 경쟁력을 키울 수 있다.