使用Kafka Stream与Twitter API同步推特至MySQL时出现SQL语法错误

我使用 Kafka Stream 和 Twitter API 获取推特数据，尝试写入 MySQL 时失败，报错信息如下：

1064, "You have an error in your SQL syntax; check the manual that corresponds to your MariaDB for the right syntax to use near 'str.' at line 1"

手动执行对应 SQL 查询可以成功，但代码执行就报错，无法定位问题。我使用的是 XAMPP v3.3.0，tweets 表包含字段：tweet（文本类型）、username（文本类型）、pnr（整数类型）、prediction（整数类型）、tweet_id（整数类型）。

相关代码如下：

def insert_tweet(tweet,username,pnr,prediction,tweet_id):
   query = "INSERT INTO tweets(tweet,username,pnr,prediction,tweet_id) VALUES ('"+tweet+"','"+username+"',"+str(pnr)+","+str(int(prediction))+","+str(tweet_id)+");"
try:
    conn = MySQLdb.connect("localhost","root","","twitter" )
    cursor = conn.cursor()
    cursor.execute(query)
    print("Database insertion SUCCESSFUL!!")
    conn.commit()
except MySQLdb.Error as e:
    print(e)
    print("Database insertion unsuccessful!!")
finally:
    conn.close()


def process_data(data):

        print("Processing data ...")        

        if (not data.isEmpty()):
            nbModel=bc_model.value
            hashingTF = HashingTF(100000)
            tf = hashingTF.transform(data.map(lambda x: x[0].encode('utf-8','ignore')))
            tf.cache()
            idf = IDF(minDocFreq=2).fit(tf)
            tfidf = idf.transform(tf)
            tfidf.cache()
            prediction=nbModel.predict(tfidf)

            temp = []
            i=0
            for p,q,r in data.collect():
                temp.append([])
                temp[i].append(p.encode('utf-8','ignore'))
                temp[i].append(q)
                temp[i].append(r)
                i+=1
            i=0
            for p in prediction.collect():
                temp[i].append(p)
                i+=1        

            print(temp)
            for i in temp:
                insert_tweet(str(i[0]),str(i[1]),"0",int(i[3]),int(i[2]))
        else:
            print("Empty RDD !!!")        
            pass

twitter=tweets.map(lambda tweet: tweet['data']['id'])
tweet_text = tweets.map(lambda tweet: tweet['data']['text'])

txt = tweets.map(lambda x: (x['data']['text'], x['data']['id'], x['data']['id']))
txt.foreachRDD(process_data)

问题原因 #

错误根源是直接拼接字符串生成 SQL 语句：

1. 推特内容中可能包含单引号（比如 I'm happy）、换行符等特殊字符，这些字符会破坏 SQL 的语法结构，导致解析错误；
2. 错误提示中的 str. 大概率是推特文本里包含了类似片段，被错误解析为 SQL 语法的一部分。

手动执行时用的是无特殊字符的测试数据，所以能成功，但代码处理真实推特数据时就会触发问题。

解决方案 #

改用参数化查询，让数据库驱动自动处理参数转义和类型匹配，同时避免 SQL 注入风险。

修改后的 insert_tweet 函数：

def insert_tweet(tweet, username, pnr, prediction, tweet_id):
    # 使用 %s 作为参数占位符，无需手动拼接字符串
    query = "INSERT INTO tweets(tweet, username, pnr, prediction, tweet_id) VALUES (%s, %s, %s, %s, %s);"
    try:
        conn = MySQLdb.connect("localhost", "root", "", "twitter")
        cursor = conn.cursor()
        # 将参数以元组形式传入，驱动自动处理转义与类型转换
        cursor.execute(query, (tweet, username, pnr, prediction, tweet_id))
        print("Database insertion SUCCESSFUL!!")
        conn.commit()
    except MySQLdb.Error as e:
        print(e)
        print("Database insertion unsuccessful!!")
    finally:
        conn.close()

同时调整 process_data 中的调用，将 pnr 参数改为整数类型（匹配数据库字段类型）：

for i in temp:
    insert_tweet(str(i[0]), str(i[1]), 0, int(i[3]), int(i[2]))

内容的提问来源于stack exchange，提问作者Len Kagamine