Remember when Egon Spengler had a radical idea? We’ve got to cross the streams, he said. It was the only way to win. Well, in our world we’re wrangling data, not ghosts, but the same concept holds true. When tackling complex data problems especially, crossing the streams is not only good, but necessary.
Individual streams of data are valuable in their own right, like when you’re handling a smaller problem or ghost. They provide data related to a particular dimension—the price of a stock, the order of a customer, the metric of a device. Analytics and applications can be served by a single stream of data, but the uses are narrow and local.
Stream crossing unveils a much grander arena of possibility, one filled with history, context, and related signals. When the heroes (you know, Spengler, Venkman and the crew) needed to defeat their biggest opponent, they joined forces—and streams—to conquer the challenge. The whole was greater than the sum of the parts. Something exponential happened.
In our community, data scientists, analysts, and developers are doing equally heroic things conquering complex data problems. When you harness all the data streams available—both batch and real-time streams—you’re empowered to address sophisticated problems. And as with Spengler and Venkman, sometimes you need a few people to bring their gear and help out.
The newest data is oftentimes the most important data, but only once a framework has been established by the historical data, and only with the awareness provided by other contemporaneous events. In other words, crossing data streams is a powerful thing.
The following code example demonstrates how Deephaven takes three different data sources and combines them…
from deephaven import kafka_consumer as ck
def get_trades_stream():
return ck.consumeToTable(
{ 'bootstrap.servers' : 'demo-kafka.c.deephaven-oss.internal:9092',
'schema.registry.url' : 'http://demo-kafka.c.deephaven-oss.internal:8081' },
'io.deephaven.crypto.kafka.TradesTopic',
key = ck.IGNORE,
value = ck.avro('io.deephaven.crypto.kafka.TradesTopic-io.deephaven.crypto.Trade'),
offsets=ck.ALL_PARTITIONS_SEEK_TO_END,
table_type='append')
crypto_stream = get_trades_stream().view(formulas = ["Timestamp", "Exchange", "Instrument", "Price", "Size"])
from deephaven import read_csv
crypto_0922 = read_csv("https://media.githubusercontent.com/media/deephaven/examples/main/CryptoCurrencyHistory/CSV/CryptoTrades_20210922.csv")
.view(formulas = ["Timestamp", "Exchange", "Instrument", "Price", "Size"])
from deephaven import ParquetTools as pt
crypto_0923 = pt.read_table("/data/large/crypto/CryptoTrades_20210923.snappy.parquet")
.view(formulas = ["Timestamp", "Exchange", "Instrument", "Price", "Size"])
from deephaven import merge
merge_3_and_agg = merge(crypto_stream, crypto_0922)
.view(formulas = ["Instrument", "Exchange", "Trade_Count = 1", "Base_Tot_Value = Price * Size"])
.sum_by(by = ["Instrument", "Exchange"])
from deephaven.DateTimeUtils import formatDate
join_3_tables = crypto_stream.update_view(formulas = ["Date = formatDate(Timestamp, TZ_NY)"]).countBy("Trade_Count", "Date", "Exchange", "Instrument")
.natural_join(table = crypto_0922.countBy(col="Trade_Count", by = ["Exchange", "Instrument"]), on =["Exchange, Instrument"], joins = ["Trade_Count_0922 = Trade_Count"])
.natural_join(table = crypto_0923.countBy(col="Trade_Count", by = ["Exchange, joins = Instrument"]), on =["Exchange, Instrument"], joins = ["Trade_Count_0923 = Trade_Count"])
Source link
lol