Блог

Я пытаюсь загрузить множество CSV-файлов с разных клиентов, которые содержат одни и те же типы данных, но с разными именами столбцов. Например

 Source | Medium | Date
Src | Med | Conversion Date
Came From | Format | DateTime
  

Все эти столбцы следует считать одинаковыми. Итак, Source, Src и Came From all должны перейти в столбец базы данных «Источник». Они могут быть названы как угодно для разных CSV-файлов и располагаться в любом порядке, поэтому при каждом создании другого клиента должно выполняться некоторое сопоставление.

Pandas имеет функцию to_sql, но для этого требуется, чтобы вы вручную вводили имена столбцов, а я не хочу кучу разных таблиц, потому что мне нужно отобразить одну и ту же таблицу для каждого клиента позже.

Одно из решений, которое я мог бы реализовать, — это заставить интерфейс требовать от администратора вручную выбирать столбцы и сопоставлять их с соответствующим именем столбца «master». Затем на серверной части просто переименуйте эти столбцы перед запуском в_sql.

Есть ли какой-либо другой способ, который был бы более эффективным для выполнения этого? Возможно, выполнение итерации по фрейму данных и обработка данных строка за строкой?

Я думаю, что лучший способ — создать таблицу для отношений (псевдоним -> целевой столбец) или config file . Вот только пример, но я думаю, вы можете понять мой подход:

 from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer

engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)
session = Session()
Base = declarative_base(bind=engine)


class ClientAlias(Base):
    # table for dynamic aliases
    __tablename__ = 'client_alias'
    id = Column(Integer, primary_key=True)
    alias = Column(String)
    target = Column(String)


class FinalTable(Base):
    # result table with standardized columns  - for all clients
    __tablename__ = 'final_table'
    id = Column(Integer, primary_key=True)
    client_id = Column(Integer)
    source = Column(String)
    medium = Column(String)


Base.metadata.create_all(engine)


def prepare_aliases():
    """
    insert default mapping:
    Src -> source, Came From -> source, Med -> medium, etc...
    """
    for target, aliases in (
        ('source', ('Source', 'Src', 'Came From'), ),
        ('medium', ('Medium', 'Med', 'Format'), ),
    ):
        for alias in aliases:
            session.add(ClientAlias(target=target, alias=alias))

    session.commit()

# insert a few records with client column aliases
prepare_aliases()

# example processing
dfs = (
    # first client with specific columns
    pd.DataFrame.from_dict({
        'client_id': (1, 1, ),
        'Source': ('Source11', 'Source12'),
        'Medium': ('Medium11', 'Medium12'),
    }),
    # second client with specific columns
    pd.DataFrame.from_dict({
        'client_id': (2, 2, ),
        'Src': ('Source12', 'Source22'),
        'Med': ('Medium12', 'Medium22'),
    }),
    # one more client with specific columns
    pd.DataFrame.from_dict({
        'client_id': (3, 3, ),
        'Came From': ('Source13', 'Source23'),
        'Format': ('Medium13', 'Medium23'),
    }),
    # etc...
)

# create columns map {Src -> source, Came From -> source, ect...}
columns = {c.alias: c.target for c in session.query(ClientAlias).all()}
for df in dfs:
    df.rename(columns=columns, inplace=True)

# union and insert into final table
df = pd.concat(dfs, sort=False, ignore_index=True)
df.to_sql(
    con=engine,
    name=FinalTable.__tablename__,
    index=False,
    if_exists='append'
)
  

Таким образом, вы можете добавить новую запись в client_alias (или в config file ), если у вас будет новый клиент или произойдут некоторые изменения. И все будет работать нормально без изменений кода и развертывания. В любом случае, это всего лишь пример — вы можете настроить его по своему усмотрению.