无法将扁平化JSON数据上传至MySQL的技术求助

报错信息 #

[1 rows x 21 columns]
Traceback (most recent call last):
  File "c:\Users\Anjan\OneDrive\Desktop\BIG_DATA\dataframe.py", line 50, in <module>
    df = pd.read_json(final_df)
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\util\_decorators.py", line 211, in wrapper
    return func(*args, **kwargs)
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\util\_decorators.py", line 331, in wrapper
    return func(*args, **kwargs)
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\io\json\_json.py", line 733, in read_json
    json_reader = JsonReader(
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\io\json\_json.py", line 818, in __init__
    data = self._get_data_from_filepath(filepath_or_buffer)
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\io\json\_json.py", line 858, in _get_data_from_filepath
    self.handles = get_handle(
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\io\common.py", line 704, in get_handle
    if _is_binary_mode(path_or_buf, mode) and "b" not in mode:
  File "C:\Users\Anjan\AppData\Roaming\Python\Python310\site-packages\pandas\io\common.py", line 1163, in _is_binary_mode
    return isinstance(handle, _get_binary_io_classes()) or "b" in getattr(
TypeError: argument of type 'method' is not iterable

原始JSON数据 #

{
  "reporting_entity_name": "medicare",
  "reporting_entity_type": "medicare",
  "plan_name": "medicaid",
  "plan_id_type": "hios",
  "plan_id": "1111111111",
  "plan_market_type": "individual",
  "last_updated_on": "2020-08-27",
  "version": "1.0.0",
  "in_network": [
    {
      "negotiation_arrangement": "ffs",
      "name": "Knee Replacement",
      "billing_code_type": "CPT",
      "billing_code_type_version": "2020",
      "billing_code": "27447",
      "description": "Arthroplasty, knee condyle and plateau, medial and lateral compartments",
      "negotiated_rates": [
        {
          "provider_groups": [
            {
              "npi": [0],
              "tin": {
                "type": "ein",
                "value": "11-1111111"
              }
            }
          ],
          "negotiated_prices": [
            {
              "negotiated_type": "negotiated",
              "negotiated_rate": 123.45,
              "expiration_date": "2022-01-01",
              "billing_class": "institutional"
            }
          ]
        }
      ]
    }
  ]
}

原有Python代码 #

import json
import pandas as pd
from sqlalchemy import create_engine
from functools import reduce
with open('new_ravi_test.json', 'r') as f:
    data = json.loads(f.read())

df_main = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network"]
).drop(columns="negotiated_rates")

df_provider = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network", "negotiated_rates", "provider_groups"]
)

df_prices = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network", "negotiated_rates", "negotiated_prices"]
)


dfs = [df_main, df_provider, df_prices]
final_df = reduce(lambda left, right: pd.merge(
    left,
    right,
    on=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
        "plan_id", "plan_market_type", "last_updated_on", "version"]
), dfs).explode("npi")

print(final_df)
engine = create_engine('mysql+pymysql://root:@localhost/json')

df = pd.read_json(final_df)

df.to_sql("test_file_01", con=engine, if_exists='replace', index=False)

扁平化后的DataFrame结构 #

negotiation_arrangement name    billing_code_type   billing_code_type_version   billing_code    description reporting_entity_name   reporting_entity_type   plan_name   plan_id_type    plan_id plan_market_type    last_updated_on version npi tin.type    tin.value   negotiated_type negotiated_rate expiration_date billing_class
0   ffs Knee Replacement    CPT 2020    27447   Arthroplasty, knee condyle and plateau, medial and lateral compartments medicare    medicare    medicaid    hios    1111111111  individual  2020-08-27  1.0.0   0   ein 11-1111111  negotiated  123.45  2022-01-01  institutional

错误原因 #

df = pd.read_json(final_df)是完全多余的操作——final_df已经是处理好的DataFrame，而pd.read_json()是用来从JSON文件/字符串生成DataFrame的，把DataFrame传进去必然触发类型错误。

修正后的完整代码 #

直接去掉多余的转换步骤，用处理好的final_df直接上传：

import json
import pandas as pd
from sqlalchemy import create_engine
from functools import reduce

# 读取JSON文件
with open('new_ravi_test.json', 'r') as f:
    data = json.loads(f.read())

# 扁平化主数据
df_main = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network"]
).drop(columns="negotiated_rates")

# 扁平化供应商数据
df_provider = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network", "negotiated_rates", "provider_groups"]
)

# 扁平化价格数据
df_prices = pd.json_normalize(
    data=data,
    meta=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
          "plan_id", "plan_market_type", "last_updated_on", "version"],
    record_path=["in_network", "negotiated_rates", "negotiated_prices"]
)

# 合并所有DataFrame
dfs = [df_main, df_provider, df_prices]
final_df = reduce(lambda left, right: pd.merge(
    left,
    right,
    on=["reporting_entity_name", "reporting_entity_type", "plan_name", "plan_id_type",
        "plan_id", "plan_market_type", "last_updated_on", "version"]
), dfs).explode("npi")

print(final_df)

# 连接MySQL并上传数据
engine = create_engine('mysql+pymysql://root:@localhost/json')
# 直接使用处理完成的final_df上传
final_df.to_sql("test_file_01", con=engine, if_exists='replace', index=False)

额外注意事项 #

• 确保环境已安装依赖：pip install pymysql sqlalchemy
• 检查MySQL连接字符串正确性：root:@localhost/json代表用户名root、无密码、目标数据库为json
• 若JSON数据量较大，建议分块处理避免内存溢出

内容的提问来源于stack exchange，提问作者BARE BEAR