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

подробности: Я имеюобразец твиттера данных из моей временной шкалы здесь, Чтобы добраться до моей отправной точки, вот код для чтения 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 имеет метод транспонирования для фреймов данных и dask.array имеет метод транспонирования для массивов. Моя цель состоит в том, чтобы сделать ту же операцию транспонирования, но на фрейме данных dask. Как бы я это сделал?

Конвертировать строки в столбцыИзменить для решения

Этот код решает исходную проблему транспонирования, очищает json-файлы Twitter, определяя столбцы, которые вы хотите сохранить, и удаляя остальные, и создает новый столбец, применяя функцию к Серии. Затем мы записываем НАМНОГО меньший, очищенный файл на диск.

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)