[python] 팬더를 사용한“대형 데이터”작업 흐름

나는 이런 방식으로 일상적으로 수십 기가 바이트의 데이터를 사용합니다. 예를 들어, 디스크를 통해 쿼리를 통해 읽고, 데이터를 만들고, 다시 추가하는 테이블이 있습니다.

그것의 가치가 읽기 워드 프로세서 및 후반이 스레드에 데이터를 저장하는 방법에 대한 몇 가지 제안.

데이터 저장 방법에 영향을 미치는 세부 사항은 다음과 같습니다.
최대한 상세하게 설명하십시오. 구조 개발을 도와 드릴 수 있습니다.

1. 데이터 크기, 행 수, 열, 열 유형; 행을 추가합니까, 아니면 열만 추가합니까?
2. 일반적인 작업은 어떻게됩니까? 예를 들어 열에 대한 쿼리를 수행하여 여러 행과 특정 열을 선택한 다음 작업 (메모리 내)을 수행하고 새 열을 만들고 저장하십시오.
   장난감 예제를 제공하면 더 구체적인 권장 사항을 제공 할 수 있습니다.
3. 그 처리 후, 당신은 무엇을합니까? 2 단계는 특별합니까, 아니면 반복 가능합니까?
4. 플랫 파일 입력 : Gb의 총 크기는 몇 개입니까? 예를 들어 레코드별로 어떻게 구성되어 있습니까? 각 필드에 다른 필드가 포함되어 있습니까, 아니면 각 파일에 모든 필드가있는 파일 당 레코드가 있습니까?
5. 기준에 따라 행의 하위 세트 (레코드)를 선택 했습니까 (예 : 필드 A> 5 인 행 선택)? 그런 다음 무언가를 수행하거나 모든 레코드가있는 필드 A, B, C를 선택하고 무언가를 수행합니까?
6. 모든 열 (그룹)을 ‘작업’하거나 보고서에만 사용할 수있는 적절한 비율이 있습니까 (예 : 데이터를 유지하고 싶지만 해당 열을 명시 적으로 가져올 필요는 없습니다. 최종 결과 시간)?

해결책

팬더가 적어도0.10.1 설치되어 있는지 확인하십시오 .

청크 단위의 반복 파일 과 여러 테이블 쿼리를 읽습니다 .

pytables는 쿼리 방식으로 행 단위로 작동하도록 최적화되었으므로 각 필드 그룹에 대한 테이블을 만듭니다. 이렇게하면 작은 테이블 그룹을 쉽게 선택할 수 있습니다 (큰 테이블에서 작동하지만이 방법으로 더 효율적입니다 … 앞으로이 제한을 해결할 수 있다고 생각합니다 … 더 직관적 임) :
(다음은 의사 코드입니다.)

import numpy as np
import pandas as pd

# create a store
store = pd.HDFStore('mystore.h5')

# this is the key to your storage:
#    this maps your fields to a specific group, and defines 
#    what you want to have as data_columns.
#    you might want to create a nice class wrapping this
#    (as you will want to have this map and its inversion)  
group_map = dict(
    A = dict(fields = ['field_1','field_2',.....], dc = ['field_1',....,'field_5']),
    B = dict(fields = ['field_10',......        ], dc = ['field_10']),
    .....
    REPORTING_ONLY = dict(fields = ['field_1000','field_1001',...], dc = []),

)

group_map_inverted = dict()
for g, v in group_map.items():
    group_map_inverted.update(dict([ (f,g) for f in v['fields'] ]))

파일 읽기 및 스토리지 생성 (본질적으로 수행 append_to_multiple) :

for f in files:
   # read in the file, additional options may be necessary here
   # the chunksize is not strictly necessary, you may be able to slurp each 
   # file into memory in which case just eliminate this part of the loop 
   # (you can also change chunksize if necessary)
   for chunk in pd.read_table(f, chunksize=50000):
       # we are going to append to each table by group
       # we are not going to create indexes at this time
       # but we *ARE* going to create (some) data_columns

       # figure out the field groupings
       for g, v in group_map.items():
             # create the frame for this group
             frame = chunk.reindex(columns = v['fields'], copy = False)    

             # append it
             store.append(g, frame, index=False, data_columns = v['dc'])

이제 파일에 모든 테이블이 있습니다 (실제로 원하는 경우 별도의 파일로 저장할 수 있습니다. 파일 이름을 group_map에 추가해야 할 수도 있지만 필요하지는 않습니다).

이것은 열을 가져오고 새로운 열을 만드는 방법입니다.

frame = store.select(group_that_I_want)
# you can optionally specify:
# columns = a list of the columns IN THAT GROUP (if you wanted to
#     select only say 3 out of the 20 columns in this sub-table)
# and a where clause if you want a subset of the rows

# do calculations on this frame
new_frame = cool_function_on_frame(frame)

# to 'add columns', create a new group (you probably want to
# limit the columns in this new_group to be only NEW ones
# (e.g. so you don't overlap from the other tables)
# add this info to the group_map
store.append(new_group, new_frame.reindex(columns = new_columns_created, copy = False), data_columns = new_columns_created)

사후 처리 준비가되면 :

# This may be a bit tricky; and depends what you are actually doing.
# I may need to modify this function to be a bit more general:
report_data = store.select_as_multiple([groups_1,groups_2,.....], where =['field_1>0', 'field_1000=foo'], selector = group_1)

data_columns 정보, 당신은 실제로 정의 할 필요가 없습니다 모든 data_columns을; 열을 기준으로 행을 하위 선택할 수 있습니다. 예를 들면 다음과 같습니다.

store.select(group, where = ['field_1000=foo', 'field_1001>0'])

최종 보고서 생성 단계에서 가장 흥미로울 수 있습니다 (실제로 데이터 열은 다른 열과 분리되어 있으므로 많이 정의하면 효율성에 다소 영향을 줄 수 있습니다).

다음을 원할 수도 있습니다.

• 필드 목록을 가져오고 groups_map에서 그룹을 찾은 다음 결과를 연결하여 결과 프레임을 얻는 함수를 만듭니다 (이것은 본질적으로 select_as_multiple의 기능입니다). 이렇게하면 구조가 매우 투명 해집니다.
• 특정 데이터 열에 대한 인덱스 (행 하위 설정이 훨씬 빨라짐)
• 압축을 활성화하십시오.

궁금한 점이 있으면 알려주세요!

답변

aCollection.insert((a[1].to_dict() for a in df.iterrows()))

pd.DataFrame(list(mongoCollection.find({'anAttribute':{'$gt':2887000, '$lt':2889000}})))

aJoinDF = pandas.DataFrame(list(mongoCollection.find({'anAttribute':{'$in':Att_Keys}})))

df = pandas.merge(df, aJoinDF, on=aKey, how='left')

collection.update({primarykey:foo},{key:change})

def transpose_table(h_in, table_path, h_out, group_name="data", group_path="/"):
    # Get a reference to the input data.
    tb = h_in.getNode(table_path)
    # Create the output group to hold the columns.
    grp = h_out.createGroup(group_path, group_name, filters=tables.Filters(complevel=1))
    for col_name in tb.colnames:
        logger.debug("Processing %s", col_name)
        # Get the data.
        col_data = tb.col(col_name)
        # Create the output array.
        arr = h_out.createCArray(grp,
                                 col_name,
                                 tables.Atom.from_dtype(col_data.dtype),
                                 col_data.shape)
        # Store the data.
        arr[:] = col_data
    h_out.flush()

def read_hdf5(hdf5_path, group_path="/data", columns=None):
    """Read a transposed data set from a HDF5 file."""
    if isinstance(hdf5_path, tables.file.File):
        hf = hdf5_path
    else:
        hf = tables.openFile(hdf5_path)

    grp = hf.getNode(group_path)
    if columns is None:
        data = [(child.name, child[:]) for child in grp]
    else:
        data = [(child.name, child[:]) for child in grp if child.name in columns]

    # Convert any float32 columns to float64 for processing.
    for i in range(len(data)):
        name, vec = data[i]
        if vec.dtype == np.float32:
            data[i] = (name, vec.astype(np.float64))

    if not isinstance(hdf5_path, tables.file.File):
        hf.close()
    return pd.DataFrame.from_items(data)