python restful api（如何调用elasticsearch中的数据）

本文目录

• 如何调用elasticsearch中的数据
• 如何用 flask 优雅的实现 restful api

如何调用elasticsearch中的数据

由于需要提升项目的搜索质量，最近研究了一下Elasticsearch，一款非常优秀的分布式搜索程序。最开始的一些笔记放到github，这里只是归纳总结一下。首先，为什么要使用Elasticsearch？最开始的时候，我们的项目仅仅使用MySQL进行简单的搜索，然后一个不能索引的like语句，直接拉低MySQL的性能。后来，我们曾考虑过sphinx，并且sphinx也在之前的项目中成功实施过，但想想现在的数据量级，多台MySQL，以及搜索服务本身HA，还有后续扩容的问题，我们觉得sphinx并不是一个最优的选择。于是自然将目光放到了Elasticsearch上面。根据官网自己的介绍，Elasticsearch是一个分布式搜索服务，提供Restful API，底层基于Lucene，采用多shard的方式保证数据安全，并且提供自动resharding的功能，加之github等大型的站点也采用 Elasticsearch作为其搜索服务，我们决定在项目中使用Elasticsearch。对于Elasticsearch，如果要在项目中使用，需要解决如下问题：索引，对于需要搜索的数据，如何建立合适的索引，还需要根据特定的语言使用不同的analyzer等。搜索，Elasticsearch提供了非常强大的搜索功能，如何写出高效的搜索语句？数据源，我们所有的数据是存放到MySQL的，MySQL是唯一数据源，如何将MySQL的数据导入到Elasticsearch？对于1和2，因为我们的数据都是从MySQL生成，index的field是固定的，主要做的工作就是根据业务场景设计好对应的mapping以及search语句就可以了，当然实际不可能这么简单，需要我们不断的调优。而对于3，则是需要一个工具将MySQL的数据导入Elasticsearch，因为我们对搜索实时性要求很高，所以需要将MySQL的增量数据实时导入，笔者唯一能想到的就是通过row based binlog来完成。而近段时间的工作，也就是实现一个MySQL增量同步到Elasticsearch的服务。LuceneElasticsearch底层是基于Lucene的，Lucene是一款优秀的搜索lib，当然，笔者以前仍然没有接触使用过。:-)Lucene关键概念：Document：用来索引和搜索的主要数据源，包含一个或者多个Field，而这些Field则包含我们跟Lucene交互的数据。Field：Document的一个组成部分，有两个部分组成，name和value。Term：不可分割的单词，搜索最小单元。Token：一个Term呈现方式，包含这个Term的内容，在文档中的起始位置，以及类型。Lucene使用Inverted index来存储term在document中位置的映射关系。譬如如下文档：Elasticsearch Server 1.0 （document 1）Mastring Elasticsearch （document 2）Apache Solr 4 Cookbook （document 3）使用inverted index存储，一个简单地映射关系：TermCountDocuemnt1.0 1 《1》 4 1 《3》 Apache 1 《3》 Cookbook 1 《3》 Elasticsearch 2 《1》.《2》 Mastering 1 《2》 Server 1 《1》 Solr 1 《3》 对于上面例子，我们首先通过分词算法将一个文档切分成一个一个的token，再得到该token与document的映射关系，并记录token出现的总次数。这样就得到了一个简单的inverted index。Elasticsearch关键概念要使用Elasticsearch，笔者认为，只需要理解几个基本概念就可以了。在数据层面，主要有：Index：Elasticsearch用来存储数据的逻辑区域，它类似于关系型数据库中的db概念。一个index可以在一个或者多个shard上面，同时一个shard也可能会有多个replicas。Document：Elasticsearch里面存储的实体数据，类似于关系数据中一个table里面的一行数据。document由多个field组成，不同的document里面同名的field一定具有相同的类型。document里面field可以重复出现，也就是一个field会有多个值，即multivalued。Document type：为了查询需要，一个index可能会有多种document，也就是document type，但需要注意，不同document里面同名的field一定要是相同类型的。Mapping：存储field的相关映射信息，不同document type会有不同的mapping。对于熟悉MySQL的童鞋，我们只需要大概认为Index就是一个db，document就是一行数据，field就是table的column，mapping就是table的定义，而document type就是一个table就可以了。Document type这个概念其实最开始也把笔者给弄糊涂了，其实它就是为了更好的查询，举个简单的例子，一个index，可能一部分数据我们想使用一种查询方式，而另一部分数据我们想使用另一种查询方式，于是就有了两种type了。不过这种情况应该在我们的项目中不会出现，所以通常一个index下面仅会有一个 type。在服务层面，主要有：Node: 一个server实例。Cluster：多个node组成cluster。Shard：数据分片，一个index可能会存在于多个shards，不同shards可能在不同nodes。Replica：shard的备份，有一个primary shard，其余的叫做replica shards。Elasticsearch之所以能动态resharding，主要在于它最开始就预先分配了多个shards（貌似是1024），然后以shard为单位进行数据迁移。这个做法其实在分布式领域非常的普遍，codis就是使用了1024个slot来进行数据迁移。因为任意一个index都可配置多个replica，通过冗余备份的方式保证了数据的安全性，同时replica也能分担读压力，类似于MySQL中的slave。Restful APIElasticsearch提供了Restful API，使用json格式，这使得它非常利于与外部交互，虽然Elasticsearch的客户端很多，但笔者仍然很容易的就写出了一个简易客户端用于项目中，再次印证了Elasticsearch的使用真心很容易。Restful的接口很简单，一个url表示一个特定的资源，譬如/blog/article/1，就表示一个index为blog，type为aritcle，id为1的document。***隐藏网址******隐藏网址******隐藏网址***# create***隐藏网址***# get***隐藏网址***# update***隐藏网址***# delete***隐藏网址***# exists***隐藏网址***索引和搜索虽然Elasticsearch能自动判断field类型并建立合适的索引，但笔者仍然推荐自己设置相关索引规则，这样才能更好为后续的搜索服务。我们通过定制mapping的方式来设置不同field的索引规则。而对于搜索，Elasticsearch提供了太多的搜索选项，就不一一概述了。索引和搜索是Elasticsearch非常重要的两个方面，直接关系到产品的搜索体验，但笔者现阶段也仅仅是大概了解了一点，后续在详细介绍。同步MySQL数据Elasticsearch是很强大，但要建立在有足量数据情况下面。我们的数据都在MySQL上面，所以如何将MySQL的数据导入Elasticsearch就是笔者最近研究的东西了。虽然现在有一些实现，譬如elasticsearch-river-jdbc，或者elasticsearch-river-mysql，但笔者并不打算使用。elasticsearch-river-jdbc的功能是很强大，但并没有很好的支持增量数据更新的问题，它需要对应的表只增不减，而这个几乎在项目中是不可能办到的。elasticsearch-river-mysql倒是做的很不错，采用了python-mysql-replication来通过binlog获取变更的数据，进行增量更新，但它貌似处理MySQL dump数据导入的问题，不过这个笔者真的好好确认一下？话说，python-mysql-replication笔者还提交过pull解决了minimal row image的问题，所以对elasticsearch-river-mysql这个项目很有好感。只是笔者决定自己写一个出来。为什么笔者决定自己写一个，不是因为笔者喜欢造轮子，主要原因在于对于这种MySQL syncer服务（增量获取MySQL数据更新到相关系统），我们不光可以用到Elasticsearch上面，而且还能用到其他服务，譬如cache上面。所以笔者其实想实现的是一个通用MySQL syncer组件，只是现在主要关注Elasticsearch罢了。项目代码在这里go-mysql-elasticsearch，现已完成第一阶段开发，内部对接测试中。go-mysql-elasticsearch的原理很简单，首先使用mysqldump获取当前MySQL的数据，然后在通过此时binlog的name和position获取增量数据。一些限制：binlog一定要变成row-based format格式，其实我们并不需要担心这种格式的binlog占用太多的硬盘空间，MySQL 5.6之后GTID模式都推荐使用row-based format了，而且通常我们都会把控SQL语句质量，不允许一次性更改过多行数据的。需要同步的table最好是innodb引擎，这样mysqldump的时候才不会阻碍写操作。需要同步的table一定要有主键，好吧，如果一个table没有主键，笔者真心会怀疑设计这个table的同学编程水平了。多列主键也是不推荐的，笔者现阶段不打算支持。一定别动态更改需要同步的table结构，Elasticsearch只能支持动态增加field，并不支持动态删除和更改field。通常来说，如果涉及到alter table，很多时候已经证明前期设计的不合理以及对于未来扩展的预估不足了。更详细的说明，等到笔者完成了go-mysql-elasticsearch的开发，并通过生产环境中测试了，再进行补充。总结最近一周，笔者花了不少时间在Elasticsearch上面，现在算是基本入门了。其实笔者觉得，对于一门不懂的技术，找一份靠谱的资料（官方文档或者入门书籍），蛋疼的对着资料敲一遍代码，不懂的再问google，最后在将其用到实际项目，这门技术就算是初步掌握了，当然精通还得在下点功夫。现在笔者只是觉得Elasticsearch很美好，上线之后铁定会有坑的，那时候只能慢慢填了。话说，笔者是不是要学习下java了，省的到时候看不懂代码就惨了。:-)

如何用 flask 优雅的实现 restful api

首先，安装Flaskpip install flask阅读这篇文章之前我假设你已经了解RESTful API的相关概念，如果不清楚，可以阅读我之前写的这篇博客[Designing a RESTful Web APIFlask是一个使用Python开发的基于Werkzeug的Web框架。Flask非常适合于开发RESTful API，因为它具有以下特点：?使用Python进行开发，Python简洁易懂?容易上手?灵活?可以部署到不同的环境?支持RESTful请求分发我一般是用curl命令进行测试，除此之外，还可以使用Chrome浏览器的postman扩展。资源首先，我创建一个完整的应用，支持响应/, /articles以及/article/:id。from flask import Flask, url_forapp = Flask(__name__)@app.route(’/’)def api_root():return ’Welcome’@app.route(’/articles’)def api_articles():return ’List of ’ + url_for(’api_articles’)@app.route(’/articles/《articleid》’)def api_article(articleid):return ’You are reading ’ + articleidif __name__ == ’__main__’:app.run()可以使用curl命令发送请求：响应结果分别如下所示：GET /WelcomeGET /articlesList of /articlesGET /articles/123You are reading 123路由中还可以使用类型定义:@app.route(’/articles/《articleid》’)上面的路由可以替换成下面的例子：@app.route(’/articles/《int:articleid》’)@app.route(’/articles/《float:articleid》’)@app.route(’/articles/《path:articleid》’)默认的类型为字符串。请求请求参数假设需要响应一个/hello请求，使用get方法，并传递参数namefrom flask import request@app.route(’/hello’)def api_hello():if ’name’ in request.args:return ’Hello ’ + request.argselse:return ’Hello John Doe’服务器会返回如下响应信息：GET /helloHello John DoeGET /hello?name=LuisHello Luis请求方法Flask支持不同的请求方法：@app.route(’/echo’, methods = )def api_echo():if request.method == ’GET’:return "ECHO: GET\n"elif request.method == ’POST’:return "ECHO: POST\n"elif request.method == ’PATCH’:return "ECHO: PACTH\n"elif request.method == ’PUT’:return "ECHO: PUT\n"elif request.method == ’DELETE’:return "ECHO: DELETE"可以使用如下命令进行测试:curl -X PATCH :5000/echo不同请求方法的响应如下：GET /echoECHO: GETPOST /ECHOECHO: POST...请求数据和请求头通常使用POST方法和PATCH方法的时候，都会发送附加的数据，这些数据的格式可能如下：普通文本(plain text)， JSON，XML，二进制文件或者用户自定义格式。Flask中使用request.headers类字典对象来获取请求头信息，使用request.data 获取请求数据，如果发送类型是application/json，则可以使用request.get_json()来获取JSON数据。from flask import json@app.route(’/messages’, methods = )def api_message():if request.headers == ’text/plain’:return "Text Message: " + request.dataelif request.headers == ’application/json’:return "JSON Message: " + json.dumps(request.json)elif request.headers == ’application/octet-stream’:f = open(’./binary’, ’wb’)f.write(request.data)f.close()return "Binary message written!"else:return "415 Unsupported Media Type ;)"使用如下命令指定请求数据类型进行测试：curl -H "Content-type: application/json" \-X POST :5000/messages -d ’{"message":"Hello Data"}’使用下面的curl命令来发送一个文件:curl -H "Content-type: application/octet-stream" \-X POST :5000/messages --data-binary @message.bin不同数据类型的响应结果如下所示：POST /messages {"message": "Hello Data"}Content-type: application/jsonJSON Message: {"message": "Hello Data"}POST /message 《message.bin》Content-type: application/octet-streamBinary message written!注意Flask可以通过request.files获取上传的文件，curl可以使用-F选项模拟上传文件的过程。响应Flask使用Response类处理响应。from flask import Response@app.route(’/hello’, methods = )def api_hello():data = {’hello’ : ’world’,’number’ : 3}js = json.dumps(data)resp = Response(js, status=200, mimetype=’application/json’)***隐藏网址***return resp使用-i选项可以获取响应信息：curl -i :5000/hello返回的响应信息如下所示：GET /helloHTTP/1.0 200 OKContent-Type: application/jsonContent-Length: 31***隐藏网址***Server: Werkzeug/0.8.2 Python/2.7.1Date: Wed, 25 Apr 2012 16:40:27 GMT{"hello": "world", "number": 3}mimetype指定了响应数据的类型。上面的过程可以使用Flask提供的一个简便方法实现：from flask import jsonify...# 将下面的代码替换成resp = Response(js, status=200, mimetype=’application/json’)# 这里的代码resp = jsonify(data)resp.status_code = 200状态码和错误处理如果成功响应的话，状态码为200。对于404错误我们可以这样处理：@app.errorhandler(404)def not_found(error=None):message = {’status’: 404,’message’: ’Not Found: ’ + request.url,}resp = jsonify(message)resp.status_code = 404return resp@app.route(’/users/《userid》’, methods = )def api_users(userid):users = {’1’:’john’, ’2’:’steve’, ’3’:’bill’}if userid in users:return jsonify({userid:users})else:return not_found()测试上面的两个URL，结果如下：GET /users/2HTTP/1.0 200 OK{"2": "steve"}GET /users/4HTTP/1.0 404 NOT FOUND{"status": 404,"message": "Not Found: :5000/users/4"}默认的Flask错误处理可以使用@error_handler修饰器进行覆盖或者使用下面的方法：app.error_handler_spec = not_found即使API不需要自定义错误信息，最好还是像上面这样做，因为Flask默认返回的错误信息是HTML格式的。认证使用下面的代码可以处理 HTTP Basic Authentication。from functools import wrapsdef check_auth(username, password):return username == ’admin’ and password == ’secret’def authenticate():message = {’message’: "Authenticate."}resp = jsonify(message)resp.status_code = 401resp.headers = ’Basic realm="Example"’return respdef requires_auth(f):@wraps(f)def decorated(*args, **kwargs):auth = request.authorizationif not auth:return authenticate()elif not check_auth(auth.username, auth.password):return authenticate()return f(*args, **kwargs)return decorated接下来只需要给路由增加@require_auth修饰器就可以在请求之前进行认证了：@app.route(’/secrets’)@requires_authdef api_hello():return "Shhh this is top secret spy stuff!"现在，如果没有通过认证的话，响应如下所示：GET /secretsHTTP/1.0 401 UNAUTHORIZEDWWW-Authenticate: Basic realm="Example"{"message": "Authenticate."}curl通过-u选项来指定HTTP basic authentication，使用-v选项打印请求头：curl -v -u "admin:secret"响应结果如下：GET /secrets Authorization: Basic YWRtaW46c2VjcmV0Shhh this is top secret spy stuff!Flask使用MultiDict来存储头部信息，为了给客户端展示不同的认证机制，可以给header添加更多的WWW-Autheticate。resp.headers = ’Basic realm="Example"’resp.headers.add(’WWW-Authenticate’, ’Bearer realm="Example"’)调试与日志通过设置debug=True来开启调试信息：app.run(debug=True)使用Python的logging模块可以设置日志信息：import loggingfile_handlewww.huashijixun.com?baiduzhidao.ogging.FileHandler(’app.log’)app.logger.addHandler(file_handler)app.logger.setLevel(logging.INFO)@app.route(’/hello’, methods = )def api_hello():app.logger.info(’informing’)app.logger.warning(’warning’)app.logger.error(’screaming bloody murder!’)return "check your logs\n"CURL 命令参考选项作用-X 指定HTTP请求方法，如POST，GET-H 指定请求头，例如Content-type:application/json-d 指定请求数据--data-binary 指定发送的文件-i 显示响应头部信息-u 指定认证用户名与密码-v 输出请求头部信息