ElasticSearch的客户端程序开发(R,perl,python)

1 R语言

1.1  R client for the Elasticsearch HTTP API

代码1

#https://github.com/ropensci/elastic
install.packages("elastic")
install.packages("devtools")
devtools::install_github("ropensci/elastic")
library('elastic')
connect(es_host = "172.28.11.167", es_user="elastic", es_pwd = "changeme", es_port = 9200)
Search(index = "dcsid-2017.08.03",size = 1)$hits$hits

代码2

#Stable version from CRAN
install.packages("elastic")
#还安装相依关系‘mime’, ‘openssl’, ‘R6’, ‘httr’, ‘curl’, ‘jsonlite’
#Development version from GitHub
install.packages("devtools")
#还安装相依关系‘memoise’, ‘whisker’, ‘digest’, ‘rstudioapi’, ‘git2r’, ‘withr’
devtools::install_github("ropensci/elastic")
一.
1.链接
connect(es_port = 9200)

2.使用x-pack
connect(es_host = "172.28.11.167", es_path = "", es_user="elastic", es_pwd = "changeme", es_port = 9200, es_transport_schema = "https")

#修改1
connect(es_host = "172.28.11.167", es_user="elastic", es_pwd = "changeme", es_port = 9200)
Search(index = "dcsid-2017.08.03",size = 1)$hits$hits

1.2  elasticsearchr

#https://cran.r-project.org/web/packages/elasticsearchr/vignettes/quick_start.html
devtools::install_github("alexioannides/elasticsearchr")
es <- elastic("http://172.28.11.167:9200", "elastic", "changeme")
for_everything <- query('{
"match_all": {}
}')

2        Perl API

2.1  安装

https://metacpan.org/pod/Search::Elasticsearch
ppm install Search::Elasticsearch
# Generating HTML for Search-Elasticsearch-5.01...done
# Generating HTML for Any-URI-Escape-0.01...done
# Generating HTML for namespace-clean-0.27...done
# Generating HTML for Moo-2.003002...done
# Generating HTML for Log-Any-1.049...done
# Generating HTML for B-Hooks-EndOfScope-0.21...done
# Generating HTML for Role-Tiny-2.000005...done
# Generating HTML for Sub-Quote-2.004000...done
# Generating HTML for Class-Method-Modifiers-2.12...done
# Generating HTML for Variable-Magic-0.61...done
# Updating files in site area...done

2.2  Demo

use Search::Elasticsearch;
# Round-robin between two nodes:
my $e1 = Search::Elasticsearch->new(
nodes => [
'192.168.11.73:9200',
'192.168.11.74:9200',
'192.168.11.75:9200',
'192.168.11.76:9200',
'192.168.11.77:9200'
]
);

# Connect to cluster at 192.168.11.73:9200, sniff all nodes and round-robin between them:
my $e = Search::Elasticsearch->new(
nodes    => '192.168.11.73:9200',
cxn_pool => 'Sniff'
);

# Index a document:
$e->index(
index   => 'my_app',
type    => 'blog_post',
id      => 1,
body    => {
title   => 'Elasticsearch clients',
content => 'Interesting content...',
date    => '2013-09-24'
}
);

# Get the document:
my $doc = $e->get(
index   => 'my_app',
type    => 'blog_post',
id      => 1
);

# Search:
my $results = $e->search(
index => 'my_app',
body  => {
query => {
match => { title => 'elasticsearch' }
}
}
);

# Cluster requests:
my $health      = $e->cluster->health;
my $state       = $e->cluster->state;

# Index requests:
$e->indices->create(index=>'my_index');
$e->indices->delete(index=>'my_index');
$e->indices->delete(index=>'my_app');

foreach $key (keys %$health)
{
print "$key=>$health->{$key}\n";
}

# foreach $key (keys %$results)
# {
#     $value=$results->{$key};
#     if (ref($value) eq 'HASH') {
#         print "$key=>$value\n";
#         print "\t***********begin********* \n";
#         foreach $subkey (keys %$value){
#             print "\t$subkey=>$value->{$subkey}\n";
#         }
#         print "\t***********end********* \n";
#     }
#     else{
#         print "$key=>$value\n";
#     }
# }

foreach $key (keys %$state)
{
$value=$state->{$key};
if (ref($value) eq 'HASH') {
print "$key=>$value\n";
print "\t***********begin********* \n";
foreach $subkey (keys %$value){
$subvalue=$value->{$subkey};
if (ref($subvalue) eq 'HASH') {
print "$subkey=>$value->{$subkey}\n";
print "\t\t***********sub_begin********* \n";
foreach $sub_subkey (keys %$subvalue){
print "\t\t$sub_subkey=>$subvalue->{$sub_subkey}\n";
}
print "\tt***********sub_end********* \n";
}
else{
print "\t$subkey=>$value->{$subkey}\n";
}
}

print "\t***********end********* \n";
}
else{
print "$key=>$value\n";
}
}

3       Python

3.1  集成python API

# Official low-level client for Elasticsearch
# elasticsearch-py
# http://elasticsearch-py.rtfd.org/ # C:\Users\unicom>pip install elasticsearch
# Successfully installed elasticsearch-5.4.0 urllib3-1.22
from datetime import datetime
from elasticsearch import Elasticsearch

# Thread safety:By default we allow urllib3 to open up to 10 connections to each node,
# if your application calls for more parallelism, use the maxsize parameter to raise the limit:

# allow up to 25 connections to each node
es = Elasticsearch([
'192.168.11.73:9200',
'192.168.11.74:9200',
'192.168.11.75:9200',
'192.168.11.76:9200',
'192.168.11.77:9200'
],
maxsize=25)

# you can specify to sniff on startup to inspect the cluster and load
# balance across all nodes
es2 = Elasticsearch([
'192.168.11.73:9200',
'192.168.11.74:9200',
'192.168.11.75:9200',
'192.168.11.76:9200',
'192.168.11.77:9200'
],
sniff_on_start=True)

# you can also sniff periodically and/or after failure:
es3 = Elasticsearch([
'192.168.11.73:9200',
'192.168.11.74:9200',
'192.168.11.75:9200',
'192.168.11.76:9200',
'192.168.11.77:9200'],
sniff_on_start=True,
sniff_on_connection_fail=True,
sniffer_timeout=60)

# SSL client authentication using client_cert and client_key

es4 = Elasticsearch(
['192.168.11.73', '192.168.11.74'],
http_auth=('user', 'secret'),
# port=443,
# use_ssl=True,
# ca_certs='/path/to/cacert.pem',
# client_cert='/path/to/client_cert.pem',
# client_key='/path/to/client_key.pem',
)

doc = {
'author': 'kimchy',
'text': 'Elasticsearch: cool. bonsai cool.',
'timestamp': datetime.now(),
}

res = es.index(index="test-index", doc_type='tweet', id=1, body=doc)
print(res['created'])

res = es.get(index="test-index", doc_type='tweet', id=1)
print(res['_source'])

es.indices.refresh(index="test-index")

res = es.search(index="test-index", body={"query": {"match_all": {}}})

print("Got %d Hits:" % res['hits']['total'])

for hit in res['hits']['hits']:
print("%(timestamp)s %(author)s: %(text)s" % hit["_source"])

es.indices.create(index='test-index', ignore=400)
# ignore 404 and 400
es.indices.delete(index='test-index', ignore=[400, 404])

3.2  自定义API的demo

#-*-coding:utf8-*-
"""
date:20170602
"""
import json
import urllib2
import urllib

class HttpElasticSearch(object):

def __init__(self, url):
self.url = url

def search_one_id(self, id_in):
"""
从es查某id
"""
fix_url = self.url + "_search"
data ={"query":{}}
data["query"]["match"] = {}
data["query"]["match"]["id"] = id_in
data = json.dumps(data)
req = urllib2.Request(fix_url, data)
req.get_method = lambda:'POST'
out = json.loads(urllib2.urlopen(req, timeout=1000).read().strip())
return out

def search_title_regx(self, title_regx):
"""
根据title的正则表达式
查出相应的id
"""
fix_url = self.url + "_search"
data = {"query":{}}
data["query"]["wildcard"] = {}
data["query"]["wildcard"]["title"] = title_regx
data["_source"] = ["id", "title"]
data = json.dumps(data)
req = urllib2.Request(fix_url, data)
req.get_method = lambda:'POST'
out = json.loads(urllib2.urlopen(req, timeout=1000).read().strip())
return out

def update_one_doc(self, id_in, update_body):
"""
update_body is a dict:
like {"viewCount":"5800"}
"""
res = obj.search_one_id(id_in)
if "hits" not in res or "hits" not in res["hits"] or "_type" not in res["hits"]["hits"][0]:
return False
type_t = str(res["hits"]["hits"][0]["_type"])
fix_url = self.url + type_t + "/" + id_in + "/_update"
data = {"doc":update_body}
data = json.dumps(data)
req = urllib2.Request(fix_url, data)
req.get_method = lambda:'POST'
out = json.loads(urllib2.urlopen(req, timeout=1000).read().strip())
if "_shards" not in out or int(out["_shards"]["successful"]) != int(out["_shards"]["total"]):
return False
return True

if __name__ == "__main__":

obj = HttpElasticSearch("host:port/你自己的索引/")
#res = obj.search_one_id("docid")
#print obj.search_title_regx("Best Surprise*")
if obj.update_one_doc("d5b39859b8f4cd0fe01f0116af01a733",{"viewCount":"5900"}):
print "hihi"

支持原创

原文，公众号：clark_blog
博客：blog.csdn.net/clark_xu

代码：https://github.com/clark99