python - 如何在 Storm 拓扑中将结果写入 MySQL？

Question

我提交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

Answer 1

我为每个虚拟机使用“pip install MySQL-python”，然后拓扑可以正常运行。