使用PySpark读取MySQL表时出现TypeError报错该如何解决？

1. 底层核心报错原因 #

你遇到的TypeError: an integer is required (got type bytes)是PySpark与当前使用的Python版本不兼容导致的，属于Spark内置cloudpickle序列化组件的已知版本匹配问题，各版本对应要求如下：

• PySpark < 3.0：仅支持Python 2.7 / 3.4~3.7
• PySpark 3.0.x：支持Python 2.7 / 3.5~3.8
• PySpark 3.1.x~3.2.x：支持Python 3.6~3.9
• PySpark 3.3.x及以上：支持Python 3.7~3.10+

解决方法：先执行pip show pyspark查询当前PySpark版本，要么将Python降级到对应支持的版本，要么升级PySpark到适配当前Python的版本，推荐使用PySpark 3.3+搭配Python 3.8/3.9的组合。

2. 代码本身的可执行性错误 #

除版本问题外，你的代码还存在多处语法、逻辑错误，即使版本匹配也无法正常读取MySQL数据：

• 缩进错误：JDBC连接构造、数据读取相关代码未缩进至if __name__ == "__main__"代码块内，运行时会在SparkSession初始化前就执行相关逻辑
• SQL拼接错误：(select * from" + table_name + ")中from后未加空格，生成的SQL为select * frommock3，不符合MySQL语法规范
• JDBC URL拼接错误：host_name + " :" + port_no中:前多了一个空格，生成的URL格式错误，无法正常连接MySQL
• 参数名错误：.option("dbtable ", mysql_select_query)中dbtable后多了冗余空格，参数无法被Spark识别生效

from pyspark.sql import SparkSession
if __name__ == "__main__":
    print("Read MySQL Table Demo - Application Started...")
    spark = SparkSession \
       .builder \
       .appName("Read MySQL Table Demo") \
       .config("spark.jars", "file:///home/amel/Downloads/mysql-connector-java-8.0.25.jar") \
       .config("spark.executor.extraClassPath", "file:///home/amel/Downloads/mysql-connector-java-8.0.25.jar") \
       .config("spark.driver.extraClassPath", "file:///home/amel/Downloads/mysql-connector-java-8.0.25.jar") \
       .enableHiveSupport() \
       .getOrCreate()


    spark.sparkContext.setLogLevel("ERROR")

    mysql_db_driver_class = "com.mysql.cj.jdbc.Driver"
    table_name = "mock3"
    host_name = "localhost"
    port_no = str(3306)
    user_name = "Amel"
    password = "Amel@-1998"
    database_name = "testDb"

    mysql_select_query = "(select * from " + table_name + ") as users"
    print("Printing mysql_select_query:")
    print(mysql_select_query)

    mysql_jdbc_url = "jdbc:mysql://" + host_name + ":" + port_no + "/" + database_name + "?useSSL=false&serverTimezone=UTC"
    print("Printing JDBC Url:" + mysql_jdbc_url)

    trans_detail_tbl_data_df = spark.read.format("jdbc") \
   .option("url", mysql_jdbc_url) \
   .option("driver", mysql_db_driver_class) \
   .option("dbtable", mysql_select_query) \
   .option("user", user_name) \
   .option("password", password) \
   .load()

    trans_detail_tbl_data_df.show(10, False)

    # 后续写入HDFS逻辑参考，替换为你自己的HDFS路径即可
    # trans_detail_tbl_data_df.write.mode("overwrite").parquet("hdfs://namenode_host:port/path/to/save/dir")

    print("Read MySQL Table Demo = Application Completed.")
    spark.stop()

注：MySQL 8.x版本驱动官方推荐使用com.mysql.cj.jdbc.Driver，URL追加时区参数避免连接报错。

内容的提问来源于stack exchange，提问作者Amel Bent