Skip to content

Instantly share code, notes, and snippets.

@Stiivi

Stiivi/bubbles_pipeline_join.py

Last active January 19, 2018 12:00
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Stiivi/5907305 to your computer and use it in GitHub Desktop.
Save Stiivi/5907305 to your computer and use it in GitHub Desktop.
Download ZIP
Demonstration of new graph-based pipeline, graph execution (see debug logs) and joins on the pipeline. Also demonstrates dynamic dispatch when the pipeline is redirected to a SQL table. Works with bubbles commit 5e108ad3a3f46580ebfe16168c58308bc914cf30 from Jul 2 2013.
Raw
bubbles_pipeline_join.py
import bubbles
stores = { "target": bubbles.open_store("sql", "sqlite:///") }
p = bubbles.Pipeline(stores=stores)
p.source_object("csv_source", resource="data.csv", infer_fields=True)
# Uncomment this and see the difference in logs - SQL will be used
# p.create("target", "data")
# Create a fork
a = p.fork()
a.aggregate("category_code", "amount")
p.aggregate(["category_code", "subcategory_code"], "amount")
# Join another fork to the pipeline
p.join_details(a, "category_code", "category_code")
p.pretty_print()
p.run()
Raw
csv_output
2013-07-02 00:18:18,018 DEBUG step 0: evaluate factory source csv_source
2013-07-02 00:18:18,028 DEBUG step 1: evaluate <bubbles.graph.CreateObjectNode object at 0x10dc6c610>
2013-07-02 00:18:18,031 DEBUG append_from: appending rows into data
2013-07-02 00:18:18,033 DEBUG step 2: evaluate operation aggregate
2013-07-02 00:18:18,033 DEBUG calling aggregate(sql)
2013-07-02 00:18:18,034 INFO called aggregate(sql)
2013-07-02 00:18:18,034 DEBUG step 3: evaluate operation aggregate
2013-07-02 00:18:18,034 DEBUG calling aggregate(sql)
2013-07-02 00:18:18,034 INFO called aggregate(sql)
2013-07-02 00:18:18,034 DEBUG step 4: evaluate operation join_details
2013-07-02 00:18:18,034 DEBUG calling join_details(sql, sql)
2013-07-02 00:18:18,035 INFO called join_details(sql, sql)
2013-07-02 00:18:18,035 DEBUG step 5: evaluate operation pretty_print
2013-07-02 00:18:18,035 DEBUG calling pretty_print(records)
+-------------+----------------+----------+------------+----------+------------+
|category_code|subcategory_code|amount_sum|record_count|amount_sum|record_count|
+-------------+----------------+----------+------------+----------+------------+
|a |da | 244691| 8| 558430| 32|
|a |dfb | 4847| 4| 558430| 32|
|a |i | 77024| 2| 558430| 32|
|a |lo | 221761| 2| 558430| 32|
|a |nn | 2325| 2| 558430| 32|
|a |oa | 5318| 6| 558430| 32|
|a |orcv | 1795| 4| 558430| 32|
|a |rcv | 347| 2| 558430| 32|
|a |s | 322| 2| 558430| 32|
|e |cs | 22983| 2| 77592| 8|
|e |da | 672| 2| 77592| 8|
|e |oe | -4726| 2| 77592| 8|
|e |re | 58663| 2| 77592| 8|
|l |b | 238617| 2| 480838| 22|
|l |dl | 226060| 8| 480838| 22|
|l |o | 65| 2| 480838| 22|
|l |ol | 12775| 8| 480838| 22|
|l |sol | 3321| 2| 480838| 22|
+-------------+----------------+----------+------------+----------+------------+
2013-07-02 00:18:18,037 INFO called pretty_print(records)
Raw
data.csv
category_code category subcategory_code subcategory line_item year amount
a Assets dfb Due from Banks Unrestricted currencies 2010 1581
a Assets dfb Due from Banks Unrestricted currencies 2009 2380
a Assets dfb Due from Banks Currencies subject to restriction 2010 222
a Assets dfb Due from Banks Currencies subject to restriction 2009 664
a Assets i Investments Trading 2010 36012
a Assets i Investments Trading 2009 41012
a Assets s Securities Securities purchased under resale agreements 2010 289
a Assets s Securities Securities purchased under resale agreements 2009 33
a Assets nn Nonnegotiable Nonnegotiable, nonintrest-bearing demand obligations on account of subscribed capital 2010 1123
a Assets nn Nonnegotiable Nonnegotiable, nonintrest-bearing demand obligations on account of subscribed capital 2009 1202
a Assets da Derivative Assets Investments 2010 13249
a Assets da Derivative Assets Investments 2009 18467
a Assets da Derivative Assets Client operations 2010 17633
a Assets da Derivative Assets Client operations 2009 19559
a Assets da Derivative Assets Borrowings 2010 87457
a Assets da Derivative Assets Borrowings 2009 82793
a Assets da Derivative Assets Other 2010 3287
a Assets da Derivative Assets Other 2009 2246
a Assets rcv Receivables Receivables to maintain value of currency holdings on account of subscribed capital 2010 171
a Assets rcv Receivables Receivables to maintain value of currency holdings on account of subscribed capital 2009 176
a Assets orcv Other Receivables Receivables from investment securities traded 2010 47
a Assets orcv Other Receivables Receivables from investment securities traded 2009 95
a Assets orcv Other Receivables Accrued income on loans 2010 764
a Assets orcv Other Receivables Accrued income on loans 2009 889
a Assets lo Loans Outstanding Net loans outstanding 2010 118104
a Assets lo Loans Outstanding Net loans outstanding 2009 103657
a Assets oa Other Assets Assets under retirement benefit plans 2010 0
a Assets oa Other Assets Assets under retirement benefit plans 2009 325
a Assets oa Other Assets Premises and equipment (net) 2010 635
a Assets oa Other Assets Premises and equipment (net) 2009 625
a Assets oa Other Assets Miscellaneous 2010 2436
a Assets oa Other Assets Miscellaneous 2009 1297
l Liabilities b Borrowings All 2010 128577
l Liabilities b Borrowings All 2009 110040
l Liabilities sol Sold or Lent Securities Sold under Repurchase Agreements, Securities Lent under Securities Lending Agreements, and Payable for Cash Collateral Received 2010 998
l Liabilities sol Sold or Lent Securities Sold under Repurchase Agreements, Securities Lent under Securities Lending Agreements, and Payable for Cash Collateral Received 2009 2323
l Liabilities dl Derivative Liabilities Investments 2010 13360
l Liabilities dl Derivative Liabilities Investments 2009 18923
l Liabilities dl Derivative Liabilities Client Operations 2010 17623
l Liabilities dl Derivative Liabilities Client Operations 2009 19551
l Liabilities dl Derivative Liabilities Borrowings 2010 78655
l Liabilities dl Derivative Liabilities Borrowings 2009 76321
l Liabilities dl Derivative Liabilities Other 2010 780
l Liabilities dl Derivative Liabilities Other 2009 847
l Liabilities o Other Payable to Maintain Value of Currency Holdings on Account of Subscribed Capital 2010 8
l Liabilities o Other Payable to Maintain Value of Currency Holdings on Account of Subscribed Capital 2009 57
l Liabilities ol Other Liabilities Payable for investment securities purchased 2010 307
l Liabilities ol Other Liabilities Payable for investment securities purchased 2009 2457
l Liabilities ol Other Liabilities Accrued charges on borrowings 2010 1190
l Liabilities ol Other Liabilities Accrued charges on borrowings 2009 1495
l Liabilities ol Other Liabilities Liabilities under retirement benefit plans 2010 1164
l Liabilities ol Other Liabilities Liabilities under retirement benefit plans 2009 662
l Liabilities ol Other Liabilities Accounts payable and misc liabilities 2010 2793
l Liabilities ol Other Liabilities Accounts payable and misc liabilities 2009 2707
e Equity cs Capital Stock Paid-in capital 2010 11492
e Equity cs Capital Stock Paid-in capital 2009 11491
e Equity da Deferred Amounts Deferred Amounts to Maintain Value of Currency Holdings 2010 313
e Equity da Deferred Amounts Deferred Amounts to Maintain Value of Currency Holdings 2009 359
e Equity re Retained Earnings Retained Earnings 2010 28793
e Equity re Retained Earnings Retained Earnings 2009 29870
e Equity oe Other Accumulated Other Comorehensive Loss 2010 -3043
e Equity oe Other Accumulated Other Comorehensive Loss 2009 -1683
Raw
interators_output
2013-07-02 00:17:06,145 DEBUG step 0: evaluate factory source csv_source
2013-07-02 00:17:06,154 DEBUG step 1: evaluate operation aggregate
2013-07-02 00:17:06,155 DEBUG retaining consumable factory source csv_source. it will be consumed 2 times
2013-07-02 00:17:06,155 DEBUG calling aggregate(rows)
2013-07-02 00:17:06,156 INFO called aggregate(rows)
2013-07-02 00:17:06,156 DEBUG step 2: evaluate operation aggregate
2013-07-02 00:17:06,156 DEBUG calling aggregate(rows)
2013-07-02 00:17:06,156 INFO called aggregate(rows)
2013-07-02 00:17:06,156 DEBUG step 3: evaluate operation join_details
2013-07-02 00:17:06,156 DEBUG calling join_details(rows, rows)
2013-07-02 00:17:06,157 INFO called join_details(rows, rows)
2013-07-02 00:17:06,157 DEBUG step 4: evaluate operation pretty_print
2013-07-02 00:17:06,157 DEBUG calling pretty_print(records)
+-------------+----------------+----------+------------+----------+------------+
|category_code|subcategory_code|amount_sum|record_count|amount_sum|record_count|
+-------------+----------------+----------+------------+----------+------------+
|a |lo | 221761| 2| 558430| 32|
|l |b | 238617| 2| 480838| 22|
|l |o | 65| 2| 480838| 22|
|a |s | 322| 2| 558430| 32|
|e |cs | 22983| 2| 77592| 8|
|a |i | 77024| 2| 558430| 32|
|a |dfb | 4847| 4| 558430| 32|
|l |sol | 3321| 2| 480838| 22|
|e |da | 672| 2| 77592| 8|
|e |re | 58663| 2| 77592| 8|
|a |rcv | 347| 2| 558430| 32|
|a |oa | 5318| 6| 558430| 32|
|a |orcv | 1795| 4| 558430| 32|
|l |ol | 12775| 8| 480838| 22|
|l |dl | 226060| 8| 480838| 22|
|e |oe | -4726| 2| 77592| 8|
|a |da | 244691| 8| 558430| 32|
|a |nn | 2325| 2| 558430| 32|
+-------------+----------------+----------+------------+----------+------------+
2013-07-02 00:17:06,157 INFO called pretty_print(records)
@ac1neepa
Copy link

After executing this code it is showing <OperationError: Operation 'source_object' not found>
what is the reason?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment