最近,我设置了一个Flask POST端点,通过Impyla模块将数据写入Impala DB。
Env: Python3.6.5在CentOS上。
Impala版本: impalad版本2.6.0-cdh5.8.0
api.py:
from flask import Flask, request, abort, Response
from flask_cors import CORS
import json
from impala.dbapi import connect
import sys
import re
from datetime import datetime
app = application = Flask(__name__)
CORS(app)
conn = connect(host='datanode2', port=21050,
user='user', database='testdb')
@app.route("/api/endpoint", methods=['POST'])
def post_data():
# if not request.json:
# abort(400)
params = request.get_json(force=True) # getting request data
print(">>>>>> ", params, flush=True)
params['log_time'] = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
# params['page_url'] = re.sub(
# '[^a-zA-Z0-9-_*.]', '', re.sub(':', '_', params['page_url']))
try:
cursor = conn.cursor()
sql = "INSERT INTO table ( page_title, page_url, log_time, machine, clicks, id ) VALUES (%s, %s, %s, %s, %s, %s)"
values = (params['page_title'], params['page_url'], params['log_time'],
params['machine'], params['clicks'], params['id'])
print(">>>>>> " + sql % values, file=sys.stderr, flush=True)
cursor.execute(sql, values)
print(
f">>>>>> Data Written Successfully", file=sys.stderr, flush=True)
return Response(json.dumps({'success': True}), 201, mimetype="application/json")
except Exception as e:
print(e, file=sys.stderr, flush=True)
return Response(json.dumps({'success': False}), 400, mimetype="application/json")
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5008, debug=True)req.py:
import requests as r
url = "http://123.234.345.456:30001/"
# url = "https://stackoverflow.com/questions/ask"
res = r.post('http://localhost:5008/api/endpoint',
json={
"page_title": "Home",
"page_url": url,
"machine": "Mac OS",
"clicks": 16,
"id": "60cd1d79-eda7-44c2-a4ec-ffdd5d6ac3db"
}
)
if res.ok:
print(res.json())
else:
print('Error!')我用python api.py运行了烧瓶api,然后用python req.py测试了它。
烧瓶服务器提供了以下错误:
>>>>>> {'page_title': 'Home', 'page_url': 'http://123.234.345.456:30001/', 'machine': 'Mac OS', 'clicks': 16, 'id': '60cd1d79-eda7-44c2-a4ec-ffdd5d6ac3db'}
>>>>>> INSERT INTO table ( page_title, page_url, log_time, machine, clicks, id ) VALUES (Home, http://123.234.345.456:30001/, 2018-12-12 16-14-04, Mac OS, 16, 60cd1d79-eda7-44c2-a4ec-ffdd5d6ac3db)
AnalysisException: Syntax error in line 1:
..., 'http://123.234.345.456'2018-12-12 16-14-04'0001/', ...
^
Encountered: INTEGER LITERAL
Expected: AND, AS, ASC, BETWEEN, CROSS, DESC, DIV, ELSE, END, FOLLOWING, FROM, FULL, GROUP, HAVING, ILIKE, IN, INNER, IREGEXP, IS, JOIN, LEFT, LIKE, LIMIT, NOT, NULLS, OFFSET, OR, ORDER, PRECEDING, RANGE, REGEXP, RIGHT, RLIKE, ROWS, THEN, UNION, WHEN, WHERE, COMMA, IDENTIFIER
CAUSED BY: Exception: Syntax error这个错误有点烦人:
那么它是某种条件字符转义的问题吗?我设法绕过这个问题,用一些正则表达式调整url (取消注释第27-28行)。但这真的很烦人,我不想为此清理我的数据。
当我检查其他人的试验时,人们认为在每个插入值中添加一对引号是有效的。但是,在使用字符串格式时,如何做到这一点,并且必须在cursor.execute(sql, values)之前进行
发布于 2018-12-14 10:51:25
在经历了一些挣扎,并从“零开始”的N‘work和@msafiullah在参数替换问题#317的巨大帮助,我设法使它工作。这有点复杂,所以我将为文档发布完整的代码:
错误原因:冒号转义通过Impyla发出。
解决方案:使用自定义的转义函数来处理数据,并采用Solution( Python的字符串格式化方法来替代参数),而不是标准的Python,例如cursor.execute(sql, values)。
api.py:
from flask import Flask, request, abort, Response
from flask_cors import CORS
import json
from impala.dbapi import connect
from impala.util import _escape
import sys
from datetime import datetime
import six
app = application = Flask(__name__)
CORS(app)
conn = connect(host='datanode2', port=21050,
user='user', database='testdb')
def parameterize(value): # by msafiullah
if value is None:
return "NULL"
elif isinstance(value, six.string_types):
return "'" + _escape(value) + "'"
else:
return str(value)
@app.route("/api/endpoint", methods=['POST'])
def post_data():
if not request.json:
abort(400)
params = request.get_json(force=True) # getting request data
print(">>>>>> ", params, flush=True)
params['log_time'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
cursor = conn.cursor()
sql = 'INSERT INTO table ( page_title, page_url, log_time, machine, clicks, id ) VALUES ( CAST({} AS VARCHAR(64)), {}, {}, CAST({} AS VARCHAR(32)) , {}, CAST({} AS VARCHAR(32)))'\
.format(parameterize(params['page_title']), parameterize(params['page_url']), parameterize(params['log_time']), parameterize(params['machine']), params['clicks'], parameterize(params['id']))
print(">>>>>> " + sql, file=sys.stderr, flush=True)
cursor.execute(sql)
print(
f">>>>>> Data Written Successfully", file=sys.stderr, flush=True)
return Response(json.dumps({'success': True}), 201, mimetype="application/json")
except Exception as e:
print(e, file=sys.stderr, flush=True)
return Response(json.dumps({'success': False}), 400, mimetype="application/json")
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5008, debug=True)req.py和问题是一样的。
table模式:
CREATE TABLE if not exists table (
id VARCHAR(36),
machine VARCHAR(32),
clicks INT,
page_title VARCHAR(64),
page_url STRING,
log_time TIMESTAMP
);烧瓶的服务器输出:
>>>>>> {'page_title': 'Home', 'page_url': 'http://123.234.345.456:30001/', 'machine': 'Mac OS', 'clicks': 16, 'id': '60cd1d79-eda7-44c2-a4ec-ffdd5d6ac3db'}
>>>>>> INSERT INTO table ( page_title, page_url, log_time, machine, clicks, id ) VALUES ( CAST('Home' AS VARCHAR(64)), 'http://123.234.345.456:30001/', '2018-12-14 17:27:29', CAST('Mac OS' AS VARCHAR(32)) , 16, CAST('60cd1d79-eda7-44c2-a4ec-ffdd5d6ac3db' AS VARCHAR(32)))
>>>>>> Data Written Successfully
127.0.0.1 - - [14/Dec/2018 17:27:29] "POST /api/endpoint HTTP/1.1" 201 -在黑斑羚壳内部,select * from table会给出:
+----------------------------------+--------+--------------+------------+----------------------------------------------------------------------+---------------------+
| id | machine | clicks | page_title | page_url | log_time |
+----------------------------------+--------+--------------+------------+----------------------------------------------------------------------+---------------------+
| 60cd1d79-eda7-44c2-a4ec-ffdd5d6a | Mac OS | 16 | Home | http://123.234.345.456:30001/ | 2018-12-14 17:27:29 |
+----------------------------------+--------+--------------+------------+----------------------------------------------------------------------+---------------------+基本上,只有数字(例如INT类型)不需要经过parameterize()清洗/转义过程。其他类型,如VARCHAR、CHAR、STRING、TIMESTAMP (因为冒号)应提前转义,以安全地通过Impyla插入。
发布于 2018-12-12 09:30:56
Impyla或其他基于impala的python库不支持参数化查询,传统的SQL就是这样做的。我遇到的唯一解决方案是,如果将insert值定义为字符串/时间戳,则使用引号对插入值进行包装。
您提到在执行查询之前使用字符串格式时如何做到这一点吗?简单地,只需应用字符串格式,然后插入格式化的值。
在您的示例中,让我们假设您的表具有以下类型定义:
CREATE TABLE table (
page_title VARCHAR(64),
page_url STRING,
log_time TIMESTAMP,
machine VARCHAR(64),
clicks INT,
id CHAR(36)
)那么插入语句将是:
sql = "INSERT INTO table ( page_title, page_url, log_time, machine, clicks, id ) VALUES ('%s', '%s', '%s', '%s', %s, '%s')" # note the single quotes around the string/timestamp types现在,由于log_time是时间戳类型,所以必须将datetime.now()格式化为yyyy-MM-dd HH:mm:ss格式。
params['log_time'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")如果您将log_time定义为字符串而不是时间戳,那么您的%Y-%m-%d %H-%M-%S格式就可以工作了。
最后,执行:
values = (params['page_title'], params['page_url'], params['log_time'],
params['machine'], params['clicks'], params['id'])
cursor.execute(sql, values)请注意,此方法仅在处理基本数据类型(如数字或字符串)时才有效。任何复杂的东西,如数组或结构都不能工作。
https://stackoverflow.com/questions/53739063
复制相似问题