如何在 Storm 拓扑中将结果写入 MySQL？答案

【问题标题】：How can I write result into MySQL in a Storm topology?如何在 Storm 拓扑中将结果写入 MySQL？
【发布时间】：2016-05-10 07:43:58
【问题描述】：

我提交了jar包，拓扑可以正常运行。但无法将结果写入 MySQL 也无法获取日志。我已经在我的数据库中创建了一个名为 result 的表。

我不知道该怎么办？

#coding=utf-8
import MySQLdb
import logging
from pyleus.storm import SimpleBolt

log = logging.getLogger('log_results')


def write_result(freqset,count):

    st = ''
    for i in freqset:
        st = st + i + ','
    sql = "select * from result where freqset = " + "'"+ st +"'"
    cur.execute(sql)
    returned_value = cur.fetchone()
    if returned_value != None:
        supnum = returned_value[1]+count
        sql1 = "update result set supnum = %s where freqset = '%s';" %(str(supnum),st)
        cur.execute(sql1)
        conn.commit()
    else:
        value = [st,count]
        sql2 = 'insert into result values(%s,%s)'
        cur.execute(sql2,value)
        conn.commit()



class LogResultsBolt(SimpleBolt):

    def process_tuple(self, tup):
        freqset, count = tup.values
        log.debug("%s: %d", freqset, count)
        write_result(freqset, count)

if __name__ == '__main__':

    logging.basicConfig(
            level=logging.DEBUG,
            filename='/tmp/results.log',
            format="%(message)s",
            filemode='a',
        )

    try:
        conn = MySQLdb.connect(host='10.1.1.5',user='root',passwd='',db='datamining',port=3306)
        cur = conn.cursor()
        LogResultsBolt().run()
        cur.close()
        conn.close()
    except MySQLdb.Error,e:
        log.debug("{0} {1}".format(e.args[0], e.args[1]))

我收到了调试信息：

11814 [Thread-12-result-count] ERROR backtype.storm.daemon.executor - 
java.lang.RuntimeException: Error when launching multilang subprocess
Traceback (most recent call last):
  File "/usr/lib64/python2.7/runpy.py", line 162, in _run_module_as_main
    "__main__", fname, loader, pkg_name)
  File "/usr/lib64/python2.7/runpy.py", line 72, in _run_code
    exec code in run_globals
  File "/tmp/e16601e5-a293-4233-bb9e-0e68090a44d6/supervisor/stormdist/association-miner-1-1462975821/resources/association-miner/result.py", line 2, in <module>
    import MySQLdb
ImportError: No module named MySQLdb

我的 topology.yaml 文件：

name: association-miner

topology:

    - spout:
        name: trans-spout
        module: association-miner.spout

    - bolt:
        name: minning
        module: association-miner.minning
        parallelism_hint: 3
        groupings:
            - shuffle_grouping: trans-spout

    - bolt:
        name: result-count
        module: association-miner.result
        groupings:
            - global_grouping: minning

我已经将“MySQL-python”写入 requirements.txt

【问题讨论】：

尝试在全局级别实例化一次：` conn = MySQLdb.connect(host='10.1.1.5',user='root',passwd='',db='datamining',port =3306) ` 当你的 Bolt 停止时关闭它。（顺便说一句，cur.execute(sql1) conn.commit
至于日志记录，我看不到您在哪里登录代码。无论哪种方式，如果您为记录器设置一个特定的绝对路径/文件来写入，您应该能够看到结果。（亲戚将无法工作，因为 Storm 的多语言内容被移动到临时文件夹以便运行 - 这让我感到困惑一段时间...）
Storm 故障排除可能有点棘手，不幸的是，我花了很多时间。我建议从日志记录开始，以便更好地了解 MySQL 问题的原因，然后从那里继续 :)。告诉我，祝你好运！
我已经提交了完整的代码，当我提交到storm时，我无法获取日志。但是其他bolts的日志是正常的。
感谢您提供更多详细信息。那么运行代码的vm是否安装了对应的MySQL包（如果我没记错的话是MySQL-python）？

标签： python mysql apache-storm pyleus

【解决方案1】：

我对每个虚拟机使用‘pip install MySQL-python’，然后Topology就可以正常运行了。

【讨论】：