Покупка оперативной памяти, чтобы избежать разбивки на файлы размером 30-50 ГБ плюс

Question

Я использую панды для чтения очень больших CSV-файлов, которые также разархивированы. Я распаковываю в csv файлы размером примерно 30-50 ГБ. Я делю файлы на части и обрабатываю / манипулирую ими. Наконец добавьте соответствующие данные в файлы HDF5, которые я сжимаю

Работает нормально, но медленно, так как приходится иметь дело с одним файлом в день и иметь данные за несколько лет (несжатый csv 600 ТБ)

Может ли покупка большего количества оперативной памяти быть хорошим способом избежать порчи и ускорить процесс, скажем, 64 ГБ /128 ГБ? Но сделает ли это панд медленными и громоздкими? Прав ли я, говоря, что переключение на C++ может ускорить процесс, но я все еще страдаю от процесса чтения и необходимости разбираться с данными порциями. Наконец, у кого-нибудь есть мысли о том, как лучше всего с этим справиться.

Между прочим, как только работа будет выполнена, мне больше не придется возвращаться и снова работать с данными, поэтому хочу просто заставить их работать в разумные сроки, написав кое-что, что параллельные процессы могут быть хорошими, но с ограниченным опытом в этой области это будет Мне нужно время, чтобы построить его, поэтому предпочитаю не делать этого, если только это не единственный вариант.

Обновить. Я думаю, что было бы легче увидеть код. Я не верю, что код особенно медленный в любом случае. Я думаю, что технология / методология может быть.

def txttohdf(path, contract):
    #create dataframes for trade and quote
    dftrade = pd.DataFrame(columns = ["datetime", "Price", "Volume"])
    dfquote = pd.DataFrame(columns = ["datetime", "BidPrice", "BidSize","AskPrice", "AskSize"])
    #create an hdf5 file with high compression and table so we can append
    hdf = pd.HDFStore(path + contract + '.h5', complevel=9, complib='blosc')
    hdf.put('trade', dftrade, format='table', data_columns=True)
    hdf.put('quote', dfquote, format='table', data_columns=True)
    #date1 = date(start).strftime('%Y%m%d')
    #date2 = date(end).strftime('%Y%m%d')
    #dd = [date1 + timedelta(days=x) for x in range((date2-date1).days + 1)]
    #walkthrough directories
    for subdir, dir, files in os.walk(path):
        for file in files:
            #check if contract has name
            #print(file)
                #create filename from directory and file 

            filename = os.path.join(subdir, file)
                #read in csv
            if filename.endswith('.gz'):

                df = pd.read_csv(gzip.open(filename),header=0,iterator=True,chunksize = 10000, low_memory =False,  names = ['RIC','Date','Time','GMTOffset','Type','ExCntrbID','LOC','Price','Volume','MarketVWAP','BuyerID','BidPrice','BidSize','NoBuyers','SellerID','AskPrice','AskSize','NoSellers','Qualifiers','SeqNo','ExchTime','BlockTrd','FloorTrd','PERatio','Yield','NewPrice','NewVol','NewSeqNo','BidYld','AskYld','ISMABidYld','ISMAAskYld','Duration','ModDurtn','BPV','AccInt','Convexity','BenchSpd','SwpSpd','AsstSwpSpd','SwapPoint','BasePrice','UpLimPrice','LoLimPrice','TheoPrice','StockPrice','ConvParity','Premium','BidImpVol','AskImpVol','ImpVol','PrimAct','SecAct','GenVal1','GenVal2','GenVal3','GenVal4','GenVal5','Crack','Top','FreightPr','1MnPft','3MnPft','PrYrPft','1YrPft','3YrPft','5YrPft','10YrPft','Repurch','Offer','Kest','CapGain','Actual','Prior','Revised','Forecast','FrcstHigh','FrcstLow','NoFrcts','TrdQteDate','QuoteTime','BidTic','TickDir','DivCode','AdjClose','PrcTTEFlag','IrgTTEFlag','PrcSubMktId','IrgSubMktId','FinStatus','DivExDate','DivPayDate','DivAmt','Open','High','Low','Last','OpenYld','HighYld','LowYld','ShortPrice','ShortVol','ShortTrdVol','ShortTurnnover','ShortWeighting','ShortLimit','AccVolume','Turnover','ImputedCls','ChangeType','OldValue','NewValue','Volatility','Strike','Premium','AucPrice','Auc Vol','MidPrice','FinEvalPrice','ProvEvalPrice','AdvancingIssues','DecliningIssues','UnchangedIssues','TotalIssues','AdvancingVolume','DecliningVolume','UnchangedVolume','TotalVolume','NewHighs','NewLows','TotalMoves','PercentageChange','AdvancingMoves','DecliningMoves','UnchangedMoves','StrongMarket','WeakMarket','ChangedMarket','MarketVolatility','OriginalDate','LoanAskVolume','LoanAskAmountTradingPrice','PercentageShortVolumeTradedVolume','PercentageShortPriceTradedPrice','ForecastNAV','PreviousDaysNAV','FinalNAV','30DayATMIVCall','60DayATMIVCall','90DayATMIVCall','30DayATMIVPut','60DayATMIVPut','90DayATMIVPut','BackgroundReference','DataSource','BidSpread','AskSpread','ContractPhysicalUnits','Miniumumquantity','NumberPhysicals','ClosingReferencePrice','ImbalanceQuantity','FarClearingPrice','NearClearingPrice','OptionAdjustedSpread','ZSpread','ConvexityPremium','ConvexityRatio','PercentageDailyReturn','InterpolatedCDSBasis','InterpolatedCDSSpread','ClosesttoMaturityCDSBasis','SettlementDate','EquityPrice','Parity','CreditSpread','Delta','InputVolatility','ImpliedVolatility','FairPrice','BondFloor','Edge','YTW','YTB','SimpleMargin','DiscountMargin','12MonthsEPS','UpperTradingLimit','LowerTradingLimit','AmountOutstanding','IssuePrice','GSpread','MiscValue','MiscValueDescription'])
                #parse date time this is quicker than doing it while we read it in
                for chunk in df:
                    chunk['datetime'] = chunk.apply(lambda row: datetime.datetime.strptime(row['Date']+ ':' + row['Time'],'%d-%b-%Y:%H:%M:%S.%f'), axis=1)
                    #df = df[~df.comment.str.contains('ALIAS')]
                #drop uneeded columns inc date and time
                    chunk = chunk.drop(['Date','Time','GMTOffset','ExCntrbID','LOC','MarketVWAP','BuyerID','NoBuyers','SellerID','NoSellers','Qualifiers','SeqNo','ExchTime','BlockTrd','FloorTrd','PERatio','Yield','NewPrice','NewVol','NewSeqNo','BidYld','AskYld','ISMABidYld','ISMAAskYld','Duration','ModDurtn','BPV','AccInt','Convexity','BenchSpd','SwpSpd','AsstSwpSpd','SwapPoint','BasePrice','UpLimPrice','LoLimPrice','TheoPrice','StockPrice','ConvParity','Premium','BidImpVol','AskImpVol','ImpVol','PrimAct','SecAct','GenVal1','GenVal2','GenVal3','GenVal4','GenVal5','Crack','Top','FreightPr','1MnPft','3MnPft','PrYrPft','1YrPft','3YrPft','5YrPft','10YrPft','Repurch','Offer','Kest','CapGain','Actual','Prior','Revised','Forecast','FrcstHigh','FrcstLow','NoFrcts','TrdQteDate','QuoteTime','BidTic','TickDir','DivCode','AdjClose','PrcTTEFlag','IrgTTEFlag','PrcSubMktId','IrgSubMktId','FinStatus','DivExDate','DivPayDate','DivAmt','Open','High','Low','Last','OpenYld','HighYld','LowYld','ShortPrice','ShortVol','ShortTrdVol','ShortTurnnover','ShortWeighting','ShortLimit','AccVolume','Turnover','ImputedCls','ChangeType','OldValue','NewValue','Volatility','Strike','Premium','AucPrice','Auc Vol','MidPrice','FinEvalPrice','ProvEvalPrice','AdvancingIssues','DecliningIssues','UnchangedIssues','TotalIssues','AdvancingVolume','DecliningVolume','UnchangedVolume','TotalVolume','NewHighs','NewLows','TotalMoves','PercentageChange','AdvancingMoves','DecliningMoves','UnchangedMoves','StrongMarket','WeakMarket','ChangedMarket','MarketVolatility','OriginalDate','LoanAskVolume','LoanAskAmountTradingPrice','PercentageShortVolumeTradedVolume','PercentageShortPriceTradedPrice','ForecastNAV','PreviousDaysNAV','FinalNAV','30DayATMIVCall','60DayATMIVCall','90DayATMIVCall','30DayATMIVPut','60DayATMIVPut','90DayATMIVPut','BackgroundReference','DataSource','BidSpread','AskSpread','ContractPhysicalUnits','Miniumumquantity','NumberPhysicals','ClosingReferencePrice','ImbalanceQuantity','FarClearingPrice','NearClearingPrice','OptionAdjustedSpread','ZSpread','ConvexityPremium','ConvexityRatio','PercentageDailyReturn','InterpolatedCDSBasis','InterpolatedCDSSpread','ClosesttoMaturityCDSBasis','SettlementDate','EquityPrice','Parity','CreditSpread','Delta','InputVolatility','ImpliedVolatility','FairPrice','BondFloor','Edge','YTW','YTB','SimpleMargin','DiscountMargin','12MonthsEPS','UpperTradingLimit','LowerTradingLimit','AmountOutstanding','IssuePrice','GSpread','MiscValue','MiscValueDescription'], axis=1)
                # convert to datetime explicitly and add nanoseconds to same time stamps
                    chunk['datetime'] = pd.to_datetime(chunk.datetime)
                #nanoseconds = df.groupby(['datetime']).cumcount()
                #df['datetime'] += np.array(nanoseconds, dtype='m8[ns]')  
                # drop empty prints and make sure all prices are valid
                    dfRic = chunk[(chunk["RIC"] == contract)]
                    if len(dfRic)>0:
                        print(dfRic)
                    if ~chunk.empty:
                        dft = dfRic[(dfRic["Type"] == "Trade")]
                        dft.dropna(subset = ["Volume"], inplace =True)
                        dft = dft.drop(["RIC","Type","BidPrice", "BidSize", "AskPrice", "AskSize"], axis=1)
                        dft = dft[(dft["Price"] > 0)]

                    # clean up bid and ask
                        dfq = dfRic[(dfRic["Type"] == "Quote")]
                        dfq.dropna(how = 'all', subset = ["BidSize","AskSize"], inplace =True)
                        dfq = dfq.drop(["RIC","Type","Price", "Volume"], axis=1)
                        dfq = dfq[(dfq["BidSize"] > 0) | (dfq["AskSize"] > 0)]
                        dfq = dfq.ffill()
                    else:
                        print("Empty")    
    #add to hdf and close if loop finished
                    hdf.append('trade', dft, format='table', data_columns=True)
                    hdf.append('quote', dfq, format='table', data_columns=True)
    hdf.close()

Answer 1

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

• сначала читайте только те колонки, которые вам действительно нужны, вместо того, чтобы читать, а затем отбрасывать их - используйте usecols=list_of_needed_columns параметр

• увеличьте ваш размер - попробуйте его с другими значениями - я бы начал с 10**5

• не использовать chunk.apply(...) для преобразования даты и времени - это очень медленно - используйте взамен pd.to_datetime(column, format='...')

• Вы можете более эффективно фильтровать данные, комбинируя несколько условий, вместо того, чтобы делать это шаг за шагом: