Как вы переносите дачный фрейм данных (преобразовывать столбцы в строки), чтобы приближаться к принципам аккуратных данных?

Question

TLDR: Я создал сигнальную карту данных из мешка для дачи. Блок данных dask обрабатывает каждое наблюдение (событие) в виде столбца. Таким образом, вместо того, чтобы иметь строки данных для каждого события, у меня есть столбец для каждого события. Цель состоит в том, чтобы перенести столбцы в строки таким же образом, что и pandas могут транспонировать фрейм данных с использованием df.T.

: У меня есть sample twitter data from my timeline here. Для того, чтобы добраться до моей отправной точки, вот код для чтения JSON с диска в dask.bag, а затем преобразовать его в dask.dataframe

import dask.bag as db 
import dask.dataframe as dd 
import json 


b = db.read_text('./sampleTwitter.json').map(json.loads) 
df = b.to_dataframe() 
df.head() 

Проблема Все мои отдельные события (т.е. твиты) записываются в виде столбцов вице-ряды. В соответствии с принципами tidy, я хотел бы иметь строки для каждого события. pandas has a transpose method for dataframes и dask.array имеет метод транспонирования для массивов. Моя цель состоит в том, чтобы выполнить ту же операцию транспонирования, но на дамбе данных. Как мне это сделать?

1. Преобразование строк в столбцах

Edit для решения

Этот код решает исходную задачу транспонирования, убирает Twitter JSon файлы, определяя столбцы, которые вы хотите сохранить и опуская остальное, и создает новый столбец, применяя функцию к серии. Затем мы пишем MUCH меньший, очищенный файл на диск.

import dask.dataframe as dd 
from dask.delayed import delayed 
import dask.bag as db 
from dask.diagnostics import ProgressBar,Profiler, ResourceProfiler, CacheProfiler 
import pandas as pd 
import json 
import glob 

# pull in all files.. 
filenames = glob.glob('~/sampleTwitter*.json') 


# df = ... # do work with dask.dataframe 
dfs = [delayed(pd.read_json)(fn, 'records') for fn in filenames] 
df = dd.from_delayed(dfs) 


# see all the fields of the dataframe 
fields = list(df.columns) 

# identify the fields we want to keep 
keepers = ['coordinates','id','user','created_at','lang'] 

# remove the fields i don't want from column list 
for f in keepers: 
    if f in fields: 
     fields.remove(f) 

# drop the fields i don't want and only keep whats necessary 
df = df.drop(fields,axis=1) 

clean = df.coordinates.apply(lambda x: (x['coordinates'][0],x['coordinates'][1]), meta= ('coords',tuple)) 
df['coords'] = clean 

# making new filenames from old filenames to save cleaned files 
import re 
newfilenames = [] 
for l in filenames: 
    newfilenames.append(re.search('(?<=\/).+?(?=\.)',l).group()+'cleaned.json') 
#newfilenames 

# custom saver function for dataframes using newfilenames 
def saver(frame,filename): 
    return frame.to_json('./'+filename) 

# converting back to a delayed object 
dfs = df.to_delayed() 
writes = [(delayed((saver)(df, fn))) for df, fn in zip(dfs, newfilenames)] 

# writing the cleaned, MUCH smaller objects back to disk 
dd.compute(*writes) 

Answer 1

Я думаю, что вы можете получить результат, который вы хотите, обходя мешок полностью, с кодом, как

import glob 

import pandas as pd 
import dask.dataframe as dd 
from dask.delayed import delayed 

filenames = glob.glob('sampleTwitter*.json') 
dfs = [delayed(pd.read_json)(fn, 'records') for fn in filenames] 
ddf = dd.from_delayed(dfs)