読者です 読者をやめる 読者になる 読者になる

Elasticsearch クラスタをマルチノードでインストールする

前回はElasticsearch を Linux上、Dockerコンテナでシングルノードでインストール起動しました。

前回のエントリは下記をご参照ください。

sfujimoto.hatenablog.com

今回も Linux上、Dockerコンテナそれぞれでのクラスタ構築を実施します。

Elasticsearch クラスタ on Linux

まずは Linux OS 上で Elasticsearch クラスタを構築します。ノード数は2台とします。

環境

Elasticsearch インストール

Elasticsearch のインストールはシングルノードと同じです。

Java 8、Elasticsearch をインストールします。

# echo "[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md" > /etc/yum.repos.d/elastic.repo

# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

# yum install -y java-1.8.0 elasticsearch
:

# rpm -q elasticsearch java-1.8.0-openjdk
elasticsearch-5.1.1-1.noarch
java-1.8.0-openjdk-1.8.0.111-2.b15.el7_3.x86_64

2台ともにインストールします。

クラスタ設定

Elasticsearch の設定をします。クラスタを組むためには以下の設定が必要です。

項目 説明
discovery.zen.ping.unicast.hosts Elasticsearch のクラスタを構成する相手のホスト名、もしくはIPアドレスを指定します。
cluster.name Elasticsearch のクラスタ名を指定します。クラスタ構成とする場合、クラスタを構成する全てのノードで名前を合わせる必要があります。指定しない場合、デフォルトの```elasticsearch```が設定されます。
node.name Elasticsearch のノード名を指定します。指定は必須ではないですが、設定する場合はクラスタ内で異なるノード名を指定します。

今回は下記のような設定としています。

ノード1
# echo "network.host: 10.255.0.100, 127.0.0.1
cluster.name: my-cluster
node.name: node1
discovery.zen.ping.unicast.hosts: 10.255.0.101" > /etc/elasticsearch/elasticsearch.yml
ノード2
# echo "network.host: 10.255.0.101, 127.0.0.1
cluster.name: my-cluster
node.name: node2
discovery.zen.ping.unicast.hosts: 10.255.0.100" > /etc/elasticsearch/elasticsearch.yml

Elasticsearch 再起動

Elasticsearch サービスを再起動します。

# systemctl restart elasticsearch

動作確認

クラスタのノードを確認します。

# curl localhost:9200/_cat/nodes
10.255.0.101 3 97 1 0.12 0.08 0.08 mdi - node2
10.255.0.100 3 97 1 0.03 0.04 0.08 mdi * node1

2つのノードが表示され、クラスタが構成されていることがわかります。

Elasticsearch コンテナ クラスタ on Docker

続いて、Docker上で Elasticsearch クラスタを構築します。今回もノードは2台です。

環境

  • Docker ホスト : Docker for Mac 1.12
  • Elasticsearch : 5.1.1

Docker コンテナ起動

Elastic 社の Github リポジトリdocker-compose.ymlが提供されていますので利用します。

$ git clone https://github.com/elastic/elasticsearch-docker.git
:

$ tree elasticsearch-docker
elasticsearch-docker
├── LICENSE
├── Makefile
├── README.md
├── build
│   ├── elasticsearch
│   │   ├── Dockerfile
│   │   ├── bin
│   │   │   └── es-docker
│   │   ├── elasticsearch.yml
│   │   └── log4j2.properties
│   └── python
│       └── Dockerfile
├── docker-compose.hostports.yml
├── docker-compose.yml
└── tests
    ├── __pycache__
    │   └── test_xpack_basic_index_crud.cpython-27-PYTEST.pyc
    ├── es_acceptance.py
    ├── es_acceptance.pyc
    ├── test_basic_index_crud.py
    ├── test_xpack_basic_index_crud.py
    └── testdata.json

6 directories, 16 files

Makefilemakeコマンド)で簡単に利用できるのですが、このまま利用するとバージョンが 5.0.0-beta1 という若干古めのバージョンです。最新を利用したいので、Makefile に記載されているバージョンを 5.1.1 に書き換えます。

$ cd elasticsearch-docker
$ sed -i 's/ELASTIC_VERSION=5.0.1/ELASTIC_VERSION=5.1.1/g' Makefile

docker-compose でクラスタを起動します。

$ make run-es-cluster
docker pull docker.elastic.co/elasticsearch/elasticsearch-alpine-base:latest
latest: Pulling from elasticsearch/elasticsearch-alpine-base
:

数分で起動します。

動作確認

クラスタのノードを確認します。

$ curl -u elastic:changeme localhost:9200/_cat/nodes
172.18.0.2  7 87 4 0.47 0.43 0.27 mdi * P7MWTJw
172.18.0.3 11 87 4 0.47 0.43 0.27 mdi - QF4Caww

2つのノードが表示され、クラスタが構成されていることがわかります。

まとめ

いかがでしたでしょうか?

データストアのクラスタリングといえば、非常に設定が面倒なイメージがありますが、Elasticsearch は非常に簡単です。インストールだけでなく、設定も簡単なのは嬉しいですね。

Python で並行処理(マルチコア、マルチスレッド)を実装する

Python のマルチコア、マルチスレッドの実装を勉強する機会がありましたので、メモ書きします。

簡単に違いを説明すると、

  • マルチコア
    • 複数のプロセスを立ち上げることで複数のCPUコアを利用することができる
    • 得意なもの:CPUバウンドな並列処理も複数のCPUコアを利用することで効率化できる
    • 苦手なもの:並列処理間でプロセスが異なり、データの受け渡しが難しい
  • マルチスレッド
    • 一つのプロセスを立ち上げ、一つのCPUコアを時分割で利用することができる
    • 得意なもの:並列処理間も同じメモリ空間を利用することができる
    • 苦手なもの:CPUバウンドな並列処理はCPUがボトルネックになり、並列処理の効果が出ない

Python による並列処理

Python はマルチコア、マルチスレッドを実装するパッケージを標準で持っています。マルチコアにはmultiprocssingパッケージ、マルチスレッドにはconcurrentパッケージを利用します。マルチスレッドはthreadingパッケージもありますが、3.2系から追加されたconcurrentパッケージが簡単だったので、今回はconcurrentパッケージで説明します。

マルチコア

まずはマルチコアの実装方法です。

同時実行数制御

マルチコアの同時実行数制御はmultiprocessing.Poolクラスの引数でプロセス数を設定するだけです。

from multiprocessing import Pool
pool = Pool(4)

こちらは 4プロセスを起動します。

並列処理実行

Pool から別プロセスで処理を実行する場合、apply_asyncメソッドを利用します。

from multiprocessing import Pool
import time

def f():
    time.sleep(1)
    print(1)

pool = Pool(4)
pool.apply_async(f)
print(2)
time.sleep(2)

こちらは実行プロセスとは別のプロセスでf関数が実行されます。普通にf関数を呼び出すだけであれば、1 -> 2 の順に表示されますが、apply_asyncで別プロセスで処理を開始しているため、2 -> 1 の順に表示されます。

python ./multi_core_test.py
2
1

非同期処理に引数を渡す

apply_asyncメソッドの第二引数にタプル形式で渡すだけです。引数が一つでも、二つでもタプルに並べるだけです。

from multiprocessing import Pool
import time

def f(num):
    time.sleep(num)
    print(num)

pool = Pool(4)
pool.apply_async(f, (1, ))
print(2)
time.sleep(2)
python ./multi_core_test.py
2
1

非同期処理の終了を待機

先ほどのサンプルで最後に 2秒のスリープを入れています。これには理由があり、別プロセスは実行プロセスの子プロセスとして起動します。そのため、実行したプロセスは処理を終えると、別プロセスで処理が実行中であろうと、プロセスを終了します。ただ今回は 1秒のスリープと print による標準出力だけなので、スリープに必要な秒数が分かりますが、複雑な処理の場合、適切なスリープの設定は難しいでしょう。そこでmultiprocessingパッケージは並列処理と同期を取ることができます。

from multiprocessing import Pool
import time

def f():
    time.sleep(1)
    print(1)

pool = Pool(4)
result = pool.apply_async(f)
print(2)
result.wait()
print(3)

apply_asyncが返すApplyResultwaitメソッドにより、並列処理の実行の終了を待つことができます。

python ./multi_core_test.py
2
1
3

非同期処理の戻り値を受け取る

先ほどのサンプルでは戻り値を返していませんでしたが、今度はf関数で返した値を取得してみましょう。

from multiprocessing import Pool
import time

def f():
    time.sleep(1)
    print(1)
    return 4

pool = Pool(4)
result = pool.apply_async(f)
print(2)
print(result.get())
print(3)

apply_asyncが返すApplyResultgetメソッドにより、並列処理の実行の終了を待ち、戻り値を取得することができます。

python ./multi_core_test.py
2
1
4
3

マルチスレッド

まずはマルチスレッドの実装方法です。

同時実行数制御

マルチスレッドの同時実行数制御はconcurrent.futures.ThreadPoolExecutorクラスの引数でプロセス数を設定するだけです。

from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(4)

こちらは 4プロセスを起動します。

並列処理実行

ThreadPoolExecutor から別スレッドで処理を実行する場合、submitメソッドを利用します。第一引数に並列実行したい関数を渡すだけです。

from concurrent.futures import ThreadPoolExecutor
import time

def f():
    time.sleep(1)
    print(1)

pool = ThreadPoolExecutor(4)
pool.submit(f)
print(2)
time.sleep(2)

こちらは別スレッドでf関数が実行されます。

python ./multi_thread_test.py
2
1

非同期処理に引数を渡す

submitメソッドの第二引数以降に引数となる値を渡すだけです。引数が一つの場合は第二引数に、引数が二つの場合は第二・第三引数に渡します。

from concurrent.futures import ThreadPoolExecutor
import time

def f(num):
    time.sleep(num)
    print(num)

pool = ThreadPoolExecutor(4)
pool.submit(f, 1)
print(2)
time.sleep(2)
python ./multi_thread_test.py
2
1

非同期処理の終了を待機

concurrent.futuresパッケージは並列処理と同期を取ることができます。as_completedを利用します。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def f(num):
    time.sleep(num)
    print(num)

pool = ThreadPoolExecutor(4)
result = pool.submit(f, 1)
print(2)
as_completed([result]).next()
print(3)
python ./multi_thread_test.py
2
1
3

非同期処理の戻り値を受け取る

先ほどのサンプルでは戻り値を返していませんでしたが、今度はf関数で返した値を取得してみましょう。

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def f(num):
    time.sleep(num)
    print(num)
    return 4

pool = ThreadPoolExecutor(4)
result = pool.submit(f, 1)
print(2)
print(as_completed([result]).next().result())
print(3)

as_completedresultメソッドで戻り値を取得することができます。

python ./multi_thread_test.py
2
1
4
3

まとめ

簡単な紹介となりましたが、標準パッケージでも様々な機能を備えています。また公式ドキュメントに分かりやすく色々な機能を紹介されていますので是非そちらもご参考ください。

http://docs.python.jp/3/library/concurrent.futures.html#module-concurrent.futures http://docs.python.jp/3.5/library/multiprocessing.html

Scrapy でページングをクローリングする

前回は Scrapy の簡単な機能、最低限の実装方法をご紹介しました。

sfujimoto.hatenablog.com

今回はページングへの対応方法をご紹介したいと思います。

前回は当ブログのトップページからエントリタイトルを取得し、表示するという Scrapy の実装をご紹介しました。ただし、あくまでトップページのみでトップページの 6件以外は表示されませんでした。今回はトップページからスクレピングした結果が次のページへのリンクがあり続ける限り、次のページのエントリタイトル一覧を取得し続ける実装方法をご紹介します。この実装方法を覚えることで、次のページに限らず、一覧ページから詳細ページへの深掘りなども実装できることとなります。

今回の目的

  • スクレイピング結果の URL からページを取得することで、動的な複数ページへのクローリング方法を学ぶ
  • 次のページや、一覧ページから詳細ページへのクローリングに対応できるようにする

概要

前回の記事で説明した通り、ページへのアクセスや、アクセス結果の引き渡しは scrapy 側で実装されています。start_urls に URL を書いたら、その URL へアクセスされ、アクセスした結果のレスポンスが parse メソッドの引数に渡されて、実行されます。start_urls に全ての URL を記載しても可能ですが、そんなのはサイトに変更がある度に手動で URL を追加、削除する必要があるため、現実的ではありません。

追加のリクエストは簡単です。parse メソッドから scrapy.Request を返すことで、引き続き リクエストを実行することができます。

試してみた

環境

実装

まずは次のページの URL を取得する XPATH を考えます。はてなブログの次のページの URL は以下となりました。

//div[@id="main-inner"]/div/span/a/@href

この XPATH から取得した URL を scrapy.Request で追加するだけです。

# -*- coding: utf-8 -*-
import scrapy


class TechSpider(scrapy.Spider):
    name = "tech"
    allowed_domains = ["sfujimoto.hatenablog.com"]
    start_urls = ['http://sfujimoto.hatenablog.com/']

    def parse(self, response):
        titles = response.xpath('//article/div/header/h1/a/text()')
        [ print(title) for title in titles.extract() ]
        
        for url in response.xpath('//div[@id="main-inner"]/div/span/a/@href').extract():
            yield scrapy.Request(url)

前回の説明から下2行を追加しただけです。print がリスト内包表記に変わっていることは気にしないでください。

指定した XPATH から次のページの URL を取得し、scrapy.Request の引数に URL を渡すことで、URL へアクセスし、その結果を同じく parse メソッドへ渡してくれます。これにより、次のページのリンクがあり続ける限りはページの取得を続けます。

動作確認

# scrapy runspider --nolog first_scrapy/spiders/tech.py
Elasticsearch をインストールして起動する
Scrapy でクローリング、スクレイピングする
Elasticsearch の概要を調べてみた
AWS CloudFormation
AWS S3
AWS ElasticCache
AWS ELB
「Amazon Web Services徹底活用ガイド」を読みました
脱vSphereClient (Apple信者向け)
AWS AMI
AWS RDS
AWS EC2
Webアプリケーション on EC2 and RDS
Fluentd(インストール編)
CentOS7(ネットワーク編)
Techブログをはじます

まとめ

Scrapy は多機能でクローリング、スクレイピングで想定される実装の多くを Scrapy で吸収してくれるため、実装が少なく済んで助かります。

Elasticsearch をシングルノードでインストールする

本日は Elasticsearch をいくつかのパターンで立ち上げる方法をご紹介します。

Elasticsearch は基本的にパッケージが用意されており、コマンド一発でインストール、コマンド一発で起動することができます。

Elasticsearch シングルノード on Linux

まずは Linux OS 上で Elasticsearch を起動します。

環境

事前準備

  • CentOS の起動までは終わっていること

Java インストール

Elasticsearch 5 系は Java 1.8系のみをサポートしています。(2016/1/3現在)Oracle Java、OpenJDK はどちらでも問題ありません。
Java をインストールします。

# yum install -y java-1.8.0
:
=============================================================================================================================================================================================================================================
 Package                                                             Arch                                           Version                                                            Repository                                       Size
=============================================================================================================================================================================================================================================
Installing:
 java-1.8.0-openjdk                                                  x86_64                                         1:1.8.0.111-2.b15.el7_3                                            updates                                         231 k
:
Installed:
  java-1.8.0-openjdk.x86_64 1:1.8.0.111-2.b15.el7_3
:
Complete!

Elasticsearch インストール

Elasticsearch は Yum、Apt といったパッケージリポジトリを提供しています。

今回は CentOS なので Yum のパッケージリポジトリを利用します。パッケージリポジトリの定義ファイルを作成します。

# echo "[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md" > /etc/yum.repos.d/elastic.repo

GPG キーをインポートします。

# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

Elasticsearch をインストールします。

# yum install -y elasticsearch
:
=============================================================================================================================================================================================================================================
 Package                                                    Arch                                                Version                                                 Repository                                                      Size
=============================================================================================================================================================================================================================================
Installing:
 elasticsearch                                              noarch                                              5.1.1-1                                                 elasticsearch-5.x                                               32 M
:
Installed:
  elasticsearch.noarch 0:5.1.1-1

Complete!

Elasticsearch を設定する

Elasticsearch のメインの設定は/etc/elasticsearch/elasticsearch.ymlというYAML形式のファイルで設定します。デフォルトでは全てコメントアウトされていて、デフォルト設定が割り当てられています。いくつか重要な設定を記載します。

項目 説明
network.host 接続を待ち受けるネットワークを設定します。デフォルトはローカルホストとなっており、自身からの接続のみを許可します。外部から接続したい場合、外部とのインタフェースのネットワークを指定します。どこからの接続からも許可する場合、```0.0.0.0```を設定します。
network.port 接続を待ち受けるポートを設定します。デフォルトは9200です。
path.data Elasticsearch のデータを保管するディレクトリパスを指定します。CentOS のデフォルトは```/var/lib/elasticsearch```です。
path.logs Elasticsearch のログを保管するディレクトリパスを指定します。CentOS のデフォルトは```/var/log/elasticsearch```です。
cluster.name Elasticsearch のクラスタ名を指定します。クラスタ構成とする場合、クラスタを構成する全てのノードで名前を合わせる必要があります。
node.name Elasticsearch のノード名を指定します。
discovery.zen.ping.unicast.hosts Elasticsearch のクラスタを構成する相手のホスト名、もしくはIPアドレスを指定します。Elasticsearch 1系まではブロードキャストでクラスタの相手を探していましたが、2系からはユニキャストとなりました。

今回は外部からの接続を許可する設定のみを行います。

# vi /etc/elasticsearch/elasticsearch.yml
network.host: 0.0.0.0

Elasticsearch を起動する

Elasticsearch を起動します。パッケージでインストールしている場合、サービスに登録されていますので、サービス起動コマンドで起動します。

# systemctl start elasticsearch

# tail /var/log/elasticsearch/elasticsearch.log
[2017-01-03T02:21:50,199][INFO ][o.e.n.Node               ] [lpOOTTv] starting ...
[2017-01-03T02:21:50,533][INFO ][o.e.t.TransportService   ] [lpOOTTv] publish_address {10.0.2.15:9300}, bound_addresses {[::]:9300}
[2017-01-03T02:21:50,538][INFO ][o.e.b.BootstrapCheck     ] [lpOOTTv] bound or publishing to a non-loopback or non-link-local address, enforcing bootstrap checks
[2017-01-03T02:21:53,786][INFO ][o.e.c.s.ClusterService   ] [lpOOTTv] new_master {lpOOTTv}{lpOOTTv4R2agyojuaC3V3g}{Imr1IzFnRMuGZ00FFOz6HA}{10.0.2.15}{10.0.2.15:9300}, reason: zen-disco-elected-as-master ([0] nodes joined)
[2017-01-03T02:21:53,864][INFO ][o.e.h.HttpServer         ] [lpOOTTv] publish_address {10.0.2.15:9200}, bound_addresses {[::]:9200}
[2017-01-03T02:21:53,864][INFO ][o.e.n.Node               ] [lpOOTTv] started

接続確認

Elasticsearch へ WebAPI を実行して接続します。

# curl localhost:9200
{
  "name" : "lpOOTTv",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "mOYwZU7STJSIMjg1P4lXLw",
  "version" : {
    "number" : "5.1.1",
    "build_hash" : "5395e21",
    "build_date" : "2016-12-06T12:36:15.409Z",
    "build_snapshot" : false,
    "lucene_version" : "6.3.0"
  },
  "tagline" : "You Know, for Search"
}

クラスタ名、ノード名、Elasticsearch、Luceneのバージョンなどが返ってきました。

インストール、起動は簡単ですね。

Elasticsearch コンテナ シングルノード on Docker

2016年11月頃から Elastic 社公式の Docker イメージの提供を開始しました。これにより、起動がより簡単になりました。

環境

  • Docker ホスト : Docker for Mac 1.12
  • Elasticsearch : 5.1.1

事前準備

  • Docker for Mac が起動していること

Elasticsearch コンテナを起動する

# docker run --rm -p 9200:9200 docker.elastic.co/elasticsearch/elasticsearch:5.1.1
Unable to find image 'docker.elastic.co/elasticsearch/elasticsearch:5.1.1' locally
5.1.1: Pulling from elasticsearch/elasticsearch
3690ec4760f9: Pull complete
f52154c3d3fc: Pull complete
4075cc5db14a: Pull complete
93889b68d3f9: Pull complete
33720f77e849: Pull complete
d1ddd1420cac: Pull complete
61d61543041b: Pull complete
4c746b62ddf3: Pull complete
2246369d1afe: Pull complete
81ea6699fce5: Pull complete
:
[2017-01-03T02:33:16,576][INFO ][o.e.n.Node               ] [rDoz9s3] starting ...
[2017-01-03T02:33:16,801][INFO ][o.e.t.TransportService   ] [rDoz9s3] publish_address {172.17.0.2:9300}, bound_addresses {[::]:9300}
[2017-01-03T02:33:16,806][INFO ][o.e.b.BootstrapCheck     ] [rDoz9s3] bound or publishing to a non-loopback or non-link-local address, enforcing bootstrap checks
[2017-01-03T02:33:19,965][INFO ][o.e.c.s.ClusterService   ] [rDoz9s3] new_master {rDoz9s3}{rDoz9s3QQQCl4LOLKun6aA}{ViB-UgFJQDy2drtkE1ry5Q}{172.17.0.2}{172.17.0.2:9300}, reason: zen-disco-elected-as-master ([0] nodes joined)
[2017-01-03T02:33:20,003][INFO ][o.e.h.HttpServer         ] [rDoz9s3] publish_address {172.17.0.2:9200}, bound_addresses {[::]:9200}
[2017-01-03T02:33:20,003][INFO ][o.e.n.Node               ] [rDoz9s3] started

起動しました。

Elasticsearch コンテナは X-Pack がインストール済みのため、認証が必須となります。デフォルトユーザーはelastic、パスワードはchangemeです。

# curl -u elastic:changeme localhost:9200
{
  "name" : "rDoz9s3",
  "cluster_name" : "docker-cluster",
  "cluster_uuid" : "ZH9Rg2RXRJag_TAEuJ9n_w",
  "version" : {
    "number" : "5.1.1",
    "build_hash" : "5395e21",
    "build_date" : "2016-12-06T12:36:15.409Z",
    "build_snapshot" : false,
    "lucene_version" : "6.3.0"
  },
  "tagline" : "You Know, for Search"
}

Elasticsearch を設定して、コンテナを起動する

このままでは Elasticsearch はデフォルト設定のままでしか起動できません。Elasticsearch を設定した状態で起動したい場合、ローカルで設定ファイルを作成して、Dockerのマウント機能を利用して、コンテナを起動します。

  • 設定ファイル
  • elasticsearch.yml
  • jvm.options
  • log4j2.properties
# vi elasticsearch-conf/elasticsearch.yml
network.host: 0.0.0.0
xpack.security.enabled: false
discovery.zen.minimum_master_nodes: 1

# vi jvm.options
-Xms2g
-Xmx2g
-XX:+UseConcMarkSweepGC
-XX:CMSInitiatingOccupancyFraction=75
-XX:+UseCMSInitiatingOccupancyOnly
-XX:+DisableExplicitGC
-XX:+AlwaysPreTouch
-server
-Xss1m
-Djava.awt.headless=true
-Dfile.encoding=UTF-8
-Djna.nosys=true
-Djdk.io.permissionsUseCanonicalPath=true
-Dio.netty.noUnsafe=true
-Dio.netty.noKeySetOptimization=true
-Dlog4j.shutdownHookEnabled=false
-Dlog4j2.disable.jmx=true
-Dlog4j.skipJansi=true
-XX:+HeapDumpOnOutOfMemoryError

# vi log4j2.properties
status = error
appender.console.type = Console
appender.console.name = console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] %marker%m%n
rootLogger.level = info
rootLogger.appenderRef.console.ref = console

# docker run --rm -p 9200:9200 -v /Users/fujimoto.shinji/elasticsearch-conf:/usr/share/elasticsearch/config/ docker.elastic.co/elasticsearch/elasticsearch:5.1.1
:
[2017-01-03T03:04:11,101][INFO ][o.e.n.Node               ] [moCi9bY] starting ...
[2017-01-03T03:04:11,355][INFO ][o.e.t.TransportService   ] [moCi9bY] publish_address {172.17.0.2:9300}, bound_addresses {[::]:9300}
[2017-01-03T03:04:11,366][INFO ][o.e.b.BootstrapCheck     ] [moCi9bY] bound or publishing to a non-loopback or non-link-local address, enforcing bootstrap checks
[2017-01-03T03:04:14,503][INFO ][o.e.c.s.ClusterService   ] [moCi9bY] new_master {moCi9bY}{moCi9bYtQ4K9aWZLwStmMw}{6KQdzI9RQ-Cqyub9A6zpLw}{172.17.0.2}{172.17.0.2:9300}, reason: zen-disco-elected-as-master ([0] nodes joined)
[2017-01-03T03:04:14,535][INFO ][o.e.h.HttpServer         ] [moCi9bY] publish_address {172.17.0.2:9200}, bound_addresses {[::]:9200}
[2017-01-03T03:04:14,537][INFO ][o.e.n.Node               ] [moCi9bY] started

起動しました。

X-Pack を無効化したので、ベーシック認証なしで接続します。

$ curl localhost:9200
{
  "name" : "moCi9bY",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "B0iUdGhNQla06J3TnvRBTA",
  "version" : {
    "number" : "5.1.1",
    "build_hash" : "5395e21",
    "build_date" : "2016-12-06T12:36:15.409Z",
    "build_snapshot" : false,
    "lucene_version" : "6.3.0"
  },
  "tagline" : "You Know, for Search"
}

無事接続できました。

まとめ

Elastic社のプロダクトはインストール、起動が非常に簡単です。on Linux でも on Docker でも簡単に起動することができました。興味はあるけど、大量のドキュメントを読むのが面倒だという人もまずは立ち上げて、触ってみてはいかがでしょうか。

次回は Elasticsearch による CRUD を試してみます。

Scrapy でクローリング、スクレイピングする

寒くなってきましたね。23日から年末年始休暇で12連休なのに何も予定がないので近所のドトールで黙々とブログ書いています。

さて、今回は先日、クローリング、スクレイピングを実装する機会があったので、その時に利用した Scrapy に関して調べた備忘録です。

Scrapy

Scrapy は Python 実装によるクローリング、スクレイピングを簡単に構築するためのフレームワークです。

HTTP リクエストや、スケジューラがフレームワークとして用意されており、利用者はフレームワークのルールに則ることでコード量少なく、クローリング、スクレイピングを実装することができます。利用者はどのページにアクセスするのか、アクセスしたページから何をどのように抽出するかを記述するだけで欲しい結果を取得することができます。

以前、requestsBeautifulSoupで実装したことがありますが、Scrapy 知れば良かった。。

今回は簡単に本ブログから記事タイトルを取得してみましょう。

対象読者

  • クローリング、スクレイピングに興味があるけど、ちゃんと調べたことがない方
  • Scrapy を触ったことがない方
  • Python は触ったことがある方

今回の目的

Scrapy を利用して本ブログから記事タイトルを抽出、出力する

環境

初期設定

Scrapy インストール

今回は pip にて Scrapy をインストールしました。

$ pip install scrapy
:
Successfully installed cryptography-1.7.1 idna-2.2 lxml-3.7.1 scrapy-1.3.0 setuptools-32.3.0 zope.interface-4.3.3

Scrapy 初期設定

Scrapy はサブコマンドで設定、実行することができます。初期設定はscrapy startprojectコマンドを実行すると、対話形式で設定値を入力することで最低限必要なファイル群を生成することができます。

$ scrapy startproject first_scrapy
New Scrapy project 'first_scrapy', using template directory 'XXXXXXXXX/.pyenv/versions/3.5.1/envs/scrapy/lib/python3.5/site-packages/scrapy/templates/project', created in:
    XXXXXXXXXXX/scrapy/first_scrapy

You can start your first spider with:
    cd first_scrapy
    scrapy genspider example example.com

$ cd first_scrapy
$ tree .
.
├── first_scrapy
│   ├── __init__.py
│   ├── __pycache__
│   │   ├── __init__.cpython-35.pyc
│   │   └── settings.cpython-35.pyc
│   ├── items.py
│   ├── middlewares.py
│   ├── pipelines.py
│   ├── settings.py
│   └── spiders
│       ├── __init__.py
│       └── __pycache__
│            └── __init__.cpython-35.pyc
└── scrapy.cfg

4 directories, 10 files

実装

Spider 実装

Spider はどのURLへアクセスするか、アクセスした結果をどのように扱うか、つまりやりたいことを実装します。最低限 Spider を実装することでクローリング、スクレイピングできます。Spider はscrapy genspiderコマンドによりテンプレートから生成できます。

$ scrapy genspider tech sfujimoto.hatenablog.com
Created spider 'tech' using template 'basic' in module:
  first_scrapy.spiders.tech

テンプレートからどのようなコードができたのか見てみましょう。

$ cat first_scrapy/spiders/tech.py
# -*- coding: utf-8 -*-
import scrapy


class TechSpider(scrapy.Spider):
    name = "tech"
    allowed_domains = ["sfujimoto.hatenablog.com"]
    start_urls = ['http://sfujimoto.hatenablog.com/']

    def parse(self, response):
        pass

allowed_domainsはクローリング先としてアクセスしてもよい FQDN を設定します。誤ったサイトへアクセスが多発したら、サイトへ迷惑をかけることになりますが、ここで指定した FQDN 以外への URL アクセスを防止することができます。クローリングはマナーが大事です。start_urlsは最初にアクセスする URL を指定します。例えば、アクセス先にページングがあり、複数ページから取得する場合、start_urlsは最初のページだけ取得し、次のページの URL を抽出することで次のページの URL にリクエストを送信することも可能です。

Scrapy を実行すると、start_urlsへHTTPリクエストした結果がresponseに渡され、parseメソッドが処理されます。なので、今回の例であれば、parseメソッドにタイトルを抽出し、出力する処理を書くだけです。

XPATH

parse メソッドを実装する前に HTML をスクレイピングする XPATH について説明します。Scrapy はいくつかのスクレイピング方法があるのですが、今回は XPATH を利用しました。XPATHXML からパスを指定することでタグや、属性、値を抽出する記法です。

例えば、以下のような XML から色々と抽出するパスの書き方を説明します。

<A>
  <B id="1" class="test">
    <C>TEXT</C>
  </B>
  <B id="2">
    <D>MESSAGE</D>
  </B>
</A>
絶対パスでTEXTを抽出する
/A/B[1]/C/text()
短縮系でTEXTを抽出する
//C/text()

※ 全てのCタグを抽出するので、Cタグが複数ある場合は同じ結果にはなりません。

属性でフィルタして、TEXTを抽出する
//B[@id="1"]/C/text()
属性でアンド条件でフィルタして、TEXTを抽出する
//B[@id="1" and @class="test"]/C/text()

簡単な例ですが、様々な抽出方法があります。

タイトルのルール(XPATH)を見つける

抽出したい文字や、URL の XPATH のパスは HTML を読み解いて、パスを探すしかありません。。私はWebブラウザに Firefox を利用していて、Firebug、FirePath を利用することで自動で XPATH を生成することができました。

ちなみに今回利用する XPATH は以下となります。

//article/div/header/h1/a/text()

実装

それでは XPATH がわかったところで、抽出したタイトルを出力する実装をしましょう。

# -*- coding: utf-8 -*-
import scrapy


class TechSpider(scrapy.Spider):
    name = "tech"
    allowed_domains = ["sfujimoto.hatenablog.com"]
    start_urls = ['http://sfujimoto.hatenablog.com/']

    def parse(self, response):
        titles = response.xpath('//article/div/header/h1/a/text()')
        for title in titles:
            print(title)

下3行を追加しただけです。簡単!

xpath の結果は extract() で展開することができます。それを print で標準出力しているだけです。

動作確認

実装が終わったら、Scrapy を実行して動作確認します。scrapy runspiderで Spider を実行することができます。

$ scrapy runspider --nolog first_scrapy/spiders/tech.py
Elasticsearch の概要を調べてみた
AWS CloudFormation
AWS S3
AWS ElasticCache
AWS ELB
「Amazon Web Services徹底活用ガイド」を読みました
脱vSphereClient (Apple信者向け)

記事タイトルが出力されました!

まとめ

いかがでしたでしょうか。たった 3行の実装で簡単にクローリング、スクレイピングを実装することができました。今回は非常に簡単な例ですが、Scrapy は複雑な要件にも対応しており多機能です。次回はもっと深掘りしたいと思います。

Elasticsearch の概要を調べてみた

今日から Elasticsearch について調べたことのメモ書きとして、当ブログにまとめていきます。

Elasticsearch とは

Elasticsearch は全文検索エンジンであり、多種多様なクエリ、集計、スケーラブル、高可用性、ユーザーフレンドリなAPI、拡張性を特徴としています。競合ツールとしては Apache Solr があげられると思います。

多種多様なクエリ

全文検索エンジンと聞いて、構文解析形態素解析を思い浮かべる方も多いかと思います。Elasticsearch でももちろん構文解析形態素解析ができます。ただそれだけではありません。

単純なところでは数値や日時の比較ができます。RDBMS でもできるので、Elasticsearch の特徴にはなりませんが、例えば、20歳以上、30歳未満といった客層データの検索、2016/1/1以降、1/31以前といった一定期間の売り上げデータの検索が可能です。

珍しいところで言えば、地理情報です。緯度経度情報を扱うことで、ある地点から、ある地点への距離を検索(算出)することができます。例えば、複数のお店の地理情報を入れておき、現在地から近い順にソートして検索することができます。

集計

Aggregation という機能でデータを集計することができます。例えば、年齢が20台、30台などといった一定のスパンでの購入件数、合計購入金額を検索でき、それにより客層の購入分析を行うことができます。例えば、1日の1時間毎の購入件数、合計購入金額を出すことでどの時間帯の売り上げが良いのか知ることができます。

スケーラブル

Elasticsearch は複数ノードでクラスタを組むことができ、データが入ってからもノードの追加・削除が可能です。運用が始まってからもパフォーマンスが不足しているようであれば、ノードを足すことができますし、パフォーマンスが余剰であれば、ノードを減らすことができます。

高可用性

クラスタで得られる恩恵はパフォーマンスの調整だけではありません。ノードでデータのレプリカを持つことで、可用性を高めることができます。全てのデータのレプリカを異なるノードに配置することで、一つのノードが障害でダウンしても、サービスの損失を防ぐことができます。データ量は倍となりますが、本番環境ではレプリカを持たせることは必須でしょう。

ユーザーフレンドリなAPI

Elasticsearch は RESTfull な API を提供しています。HTTP/HTTPS リクエストが送信できる環境であれば、誰でも利用することができます。RESTfull API だけでなく、各種言語による SDK も提供していますので、プログラムに組み込む時は SDK を利用することでエラー処理や、共通処理を実装することなく利用することができます。

拡張性

Elasticsearch はプラグインを提供しています。Elastic 社が提供するプラグインから、コミュニティ、個人が提供するプラグインなど色々あります。

Elasticsearch 関連プロダクト

Elastic社は Elasticsearch だけでなく、関連プロダクトを提供しています。Elasticsearch 5.0 から Elasticsearch を含む主要 4つのプロダクトを Elastic Stack として提供しています。4つとも OSS で Elastic社が持つ Github リポジトリで管理されています。

  • Elasticsearch
  • Kibana
  • Logstash
  • Beats

Kibana

Kibana は Elasticsearch に入っているデータを可視化するツールです。円グラフ、線グラフ、棒グラフ、地図など各種グラフにて Elasticsearch の検索、集計を利用して可視化することができます。また各種グラフを一つの画面に並べることでダッシュボードを作成することも可能です。例えば、ECサイトマーケティング担当向けにダッシュボードを作成するとしたら、アクセス数、購入金額、商品別の購入数、時間帯別の購入数と商品品目、時間帯別の購入数と購入年齢層・性別などをいい感じに可視化することができます。

また X-Pack というプラグインなどと組み合わせることで、Elasticsearchクラスタの監視、Elasticsearchへのアクセス制御なども可能となっています。

Logstash

Logstash は Elasticsearch にデータを入れる仕組みを提供するツールです。Elasticsearch は RESTfull APISDK などを利用してデータを入れることができますが定期的に入れる場合、スクリプト、プログラムの実装が必要となります。Logstash は入れたいデータの元、データを入れるElasticsearchの情報を設定ファイルに指定するだけで Elasticsearch にデータを入れることができます。例えば、Apacheaccess_log を Elasticsearch に入れる場合、ファイルプラグインaccess_log の差分を定期的に取得し、Grokプラグインaccess_log のメッセージを解析・変換し、Elasticsearch にデータを入れてくれます。それを Kibana で可視化することでアクセス解析、パフォーマンス解析することが可能です。

Logstash はプラグインの組み合わせで処理パイプラインを形成します。プラグインは大きく、インプット、フィルター、アウトプットに分けられます。インプットは取得するデータソースを表し、ファイル、JDBC、標準入力、AWS S3など多数あります。フィルターはインプットで取得したデータの絞り込み、変換を表し、文字列→日付変換、文字列をデータパターンから構造化データへの変換など多数あります。アウトプットはデータの出力先を表し、Elasticsearch だけではなく、ファイル、JDBC、標準出力など多数あります。

Beats

Beats は役割は Logstash と同じく Elasticsearch にデータを入れる仕組みを提供するツールです。Logstash との違いは、機能が限定されているが、軽量・高パフォーマンスだという点です。Logstash のような多種多様なプラグインはなく、データ取得元毎にプロダクトが別れています。データ出力先も Elasticsearch、Logstash、Apache Kafkaなどと限定的です。

現在提供されている Beats プロダクトは以下となります。

  • Packetbeat
  • Metricbeat
  • Filebeat
  • Winlogbeat

まとめ

Elasticsearch は単純な検索エンジンだけではなく、関連プロダクトと組み合わせることでユースケースが非常に広がりそうだと思っています。

次回は Elasticsearch クラスタのインストール、立ち上げ方を解説します。

AWS CloudFormation

AWSにサインアップした時、既にVPC(デフォルトVPC)やサブネット(デフォルトサブネット)が用意されています。
私はVPCから用意したかったのでデフォルトは削除し、VPC作成から行いました。
ただデフォルトVPCは唯一の存在であり、
自分で作成したVPCをデフォルトVPCには設定できません。
削除したデフォルトVPCを復元したい場合、
AWSサポートに問い合わせる必要があります。
昨日、AWSサポート(BASICサポートで対応いただけます)で復元を依頼したところ、
わずか、15分で対応いただけました。
過去、色々な有償プロダクトのサポート問い合わせを行ってきましたが、
無償サポートでこの対応の早さは素晴らしいです。

さて、今回はCloudFormationです。
私は単調作業が嫌いです。
単調作業があればどうすれば自動化できるかを考えます。

AWSはそんな私にピッタリな自動化のソリューションをいくつか用意されています。

まだ違いがよくわかっていませんが、以下のようなソリューションがあります。

  • CloudFormation
  • Elastic Beanstalk
  • OpsWorks
  • CodeDeploy

CloudFormationはAWSリソースをTemplateというJSON形式で表すことにより、
AWS ManagementConsoleからポチポチするクリック動作を自動で行ってくれる素晴らしい機能です。

それ以外のソリューションはまた別の機会で。

まずはもっとも簡単な例としてEC2インスタンスを一つ作成します。

サンプル
前提条件

  AMI IDはLaunch InstanceのAMI選択画面からコミュニティAMI画面を開き、
  AMIの名前の右にami-xxxxxxxx形式で記載されています。

  • キーペアにtest1を作成済み
  • サブネットを作成済み(ID指定)
  • セキュリティグループを作成済み(ID指定)
{
  "Resources": {
    "Ec2Instance": {
      "Type": "AWS::EC2::Instance",
      "Properties": {
        "KeyName": "test1",
        "ImageId": "ami-cbf90ecb",
        "InstanceType": "t2.micro",
        "NetworkInterfaces": [
          {
            "AssociatePublicIpAddress": "true",
            "SubnetId": "subnet-2d359d5a",
            "DeviceIndex": "0",
            "GroupSet": [
              "sg-5fae323a"
            ]
          }
        ]
      }
    }
  }
}

このテンプレートからスタックを作成すると、
EC2インスタンスが一つ作成されます。

一度、EC2を作成したことがある方はGUIで指定した覚えがある項目ばかりですよね。
そういうことです。
GUIの設定項目をテキストで外出ししておくだけです。

でも、それだけだとありがたみないは少ないですよね。
それどころか手入力なので逆に時間かかるだけですよね。
CloudFormationが生きるところは、

  • リソースの連携
  • リージョン間の複製
  • 特にBlue-Green Deployment

にあるように感じました。

リソースの連携

サンプルでは一つのリソースしか作成していませんが、
例えば前提条件が何もなかったとします。
その場合、EC2インスタンスを作成する前に
サブネット、キーペア、セキュリティグループを作成する必要があります。
これらもCloudFormationのリソースで作成することが可能です。
Resourcesの中にそれぞれ指定することで作成でき、
また他のリソースから参照指定することでIDを記載せずとも紐づけることが可能です。
GUIで作成ウィザードの中で表示されるプルダウンに表示されるのはIDだけで選択を誤ったことがあるのは私だけではないはず。

リージョン間の複製

複数リージョンに同一構成を作成するケースがあります。
そういうケースではテンプレートを使いまわすことで同一構成が作成できます。
勘がよい方はAMI IDが異なることを気になったはず。
そういう違いを吸収する仕組みがCloudFormationではJSON形式で記載できる組み込み関数を用意してくれています。
例えば、AMI IDのケースではMappings定義とFn::FindInMap関数を組み合わせることで、
リージョンに応じたAMI IDの紐付けを行うことができます。

Blue-Green Deployment

DevOpsで話題のリリース方式です。詳細はググってください。
Blue-Green Deploymentは同一構成をリリースの度に構築するため、
一つテンプレートを作っておけば、
それを元にスタックを作成するだけで同一構成を用意できます。
またスタックを削除するとテンプレートによって作成されたリソースが合わせて削除されます。
削除忘れにより無駄なコスト削減も実現できます。

いかがでしょうか。

CloudFormation使いたくなってきましたよね。
今回の私のサンプルは使い物にならないものですが、
AWSが利用用途に応じたサンプルをいくつか用意してくれています。
そちらを参考にすれば学習コスト少なく、扱うことができます。

ただいくつか不満があります。

  • デバッグができない
    • 予約語や変数の結果からどういった値が得られるのか分からない
    • 事前チェックがなく、処理が依存関係に応じて順に行われるので、後半にエラーが発生するため息が出る

この辺は経験で吸収するしかないのでしょうか。。
とりあえず色々と触ってみます。