建站
咨詢
售后
建站咨詢
新聞中心
譯者 | 楊曉娟
審校 | 梁策 孫淑娟
摘要
SingleStore 是一個(gè)非常通用的數(shù)據(jù)庫系統(tǒng)。它基于關(guān)系型技術(shù)，支持多模型功能，如鍵值、JSON、全文搜索、地理空間和時(shí)間序列。
本文將使用Kaggle的歷史 S&P 500 股票數(shù)據(jù)來探索 SingleStore 對(duì)時(shí)間序列數(shù)據(jù)的支持。我們還將構(gòu)建一個(gè)快速儀表板，使用 Streamlit 可視化燭臺(tái)圖。
本文中使用的 SQL 腳本、Python 代碼和筆記本文件可在GitHub 上獲得，支持 DBC、HTML 和 iPython 格式。
介紹
自關(guān)系數(shù)據(jù)庫技術(shù)出現(xiàn)以來，許多管理數(shù)據(jù)的新需求應(yīng)運(yùn)而生。馬丁·福勒(Martin Fowler) 等知名人士提出了混合持久化(Polyglot Persistence)作為管理各種數(shù)據(jù)和數(shù)據(jù)處理需求的一種解決方案，如圖 1 所示。
然而，混合持久化是有代價(jià)的，并招致了非議，比如：
在一篇經(jīng)常被引用的混合持久化帖子中，馬丁·福勒為一家虛構(gòu)的零售商繪制了一個(gè) Web 應(yīng)用程序，該應(yīng)用程序使用 Riak、Neo4j、MongoDB、Cassandra和一個(gè) RDBMS 來處理不同的數(shù)據(jù)集。不難想象，他的零售商的 DevOps 工程師會(huì)一個(gè)接一個(gè)地辭職。
? —斯蒂芬·皮門特爾(Stephen Pimentel)
此外：
我過去曾看到，如果你嘗試采用其中的六種[技術(shù)]，你至少需要 18 名員工來操作存儲(chǔ)端——就是說，六種存儲(chǔ)技術(shù)。那樣是不可擴(kuò)展的，而且成本太高。
? —大衛(wèi)·麥克羅里( Dave McCrory)
近年來，也有一些使用微服務(wù)來實(shí)現(xiàn)混合持久化架構(gòu)的建議。但是，SingleStore 可以通過在單個(gè)多模型數(shù)據(jù)庫系統(tǒng)中支持不同的數(shù)據(jù)類型和處理需求來提供更簡(jiǎn)單的解決方案。這帶來了許多好處，例如更低的 TCO(總擁有成本)、開發(fā)人員學(xué)習(xí)多種產(chǎn)品的負(fù)擔(dān)更少、沒有集成的麻煩等等。我們將在一系列文章中更詳細(xì)地討論 SingleStore 的多模型功能，現(xiàn)在則從時(shí)間序列數(shù)據(jù)開始。
首先，我們需要在 SingleStore 網(wǎng)站上創(chuàng)建一個(gè)免費(fèi)托管服務(wù)帳戶，并在 Databricks 網(wǎng)站上創(chuàng)建一個(gè)免費(fèi)社區(qū)版(CE)帳戶。 在撰寫本文時(shí)，SingleStore 的托管服務(wù)帳戶附帶 500 美元的積分，這對(duì)于本文中描述的案例研究來說綽綽有余。對(duì)于 Databricks CE，我們不要注冊(cè)試用版而是注冊(cè)免費(fèi)帳戶。在之前的文章中，我們指出 Spark 非常適合使用 SingleStore 進(jìn)行 ETL，所以這也是此處使用 Spark的原因。
如果你沒有 Kaggle 帳戶，請(qǐng)創(chuàng)建一個(gè)并下載all_stocks_5yr.csv文件。 Kaggle 網(wǎng)站聲明該文件大小為 29.58 MB。數(shù)據(jù)集由以下字段組成：
- date：從2013年2月8日到2018年2月7日的五年每日期間。沒有缺失值。
- open：開盤價(jià)。11個(gè)缺失值。
- high：最高價(jià)。8個(gè)缺失值。
- low：最低價(jià)。8個(gè)缺失值。
- close：收盤價(jià)。沒有缺失值。
- volume：成交量。沒有缺失值。
- name：交易代碼。505個(gè)唯一值。沒有缺失值。
在開始階段，我們會(huì)用到date、close和name信息。
配置 Databricks CE
上一篇文章給出了有關(guān)如何配置 Databricks CE以及和 SingleStore 一起使用的詳細(xì)說明，我們可以在這個(gè)用例中使用它們。
上傳 CSV 文件
要使用CSV文件，我們需要將其上傳到 Databricks CE 環(huán)境。上一篇文章提供了有關(guān)如何上傳CSV文件的詳細(xì)說明，我們可以在這個(gè)用例中使用它們。
創(chuàng)建數(shù)據(jù)庫表
在我們的SingleStore托管服務(wù)帳戶中，使用 SQL 編輯器新建一個(gè)timeseries_db數(shù)據(jù)庫。如下所示：
SQL:
CREATE DATABASE IF NOT EXISTS timeseries_db;
再創(chuàng)建一個(gè)表，如下所示：
SQL:
USE timeseries_db;
CREATE ROWSTORE TABLE IF NOT EXISTS tick (
ts DATETIME SERIES TIMESTAMP,
symbol VARCHAR(5),
price NUMERIC(18, 4),
KEY(ts)
);
每行有一個(gè)叫作 ts的時(shí)間值屬性。我們使用DATETIME而不是DATETIME(6)，因?yàn)樵诒纠形覀儾皇褂眯?shù)秒。SERIES TIMESTAMP將表列指定為默認(rèn)時(shí)間戳。在ts上創(chuàng)建一個(gè)KEY，因?yàn)檫@能讓我們高效地篩選值的范圍。
填寫筆記本
現(xiàn)在新建一個(gè) Databricks CE Python 筆記本，名為Data Loader for Time Series。把新筆記本附加到 Spark 集群上。
在一個(gè)新的代碼單元中，添加以下代碼：
Python:
from pyspark.sql.types import *
tick_schema = StructType([
StructField("ts", TimestampType(), True),
StructField("open", DoubleType(), True),
StructField("high", DoubleType(), True),
StructField("low", DoubleType(), True),
StructField("price", DoubleType(), True),
StructField("volume", IntegerType(), True),
StructField("symbol", StringType(), True)
])
此模式確保我們有正確的列類型。
在下一個(gè)代碼單元格中新建一個(gè) Dataframe，如下所示：
Python:
tick_df = spark.read.csv("/FileStore/all_stocks_5yr.csv",
header = True,
schema = tick_schema)
這會(huì)讀取CSV文件并創(chuàng)建一個(gè)名為tick_df的Dataframe。 我們還告訴Spark有一個(gè)標(biāo)題行，并要求它使用前面定義的模式。
在下一個(gè)代碼單元中，我們獲取行數(shù)：
Python:
tick_df.count()
執(zhí)行此操作，得到數(shù)值 619040。
根據(jù)先前的初步分析決定，我們刪除掉一些列，如下所示：
Python:
tick_df = tick_df.drop("open", "high", "low", "volume")
并對(duì)數(shù)據(jù)進(jìn)行排序：
Python:
tick_df = tick_df.sort("ts", "symbol")
在下一個(gè)代碼單元中，我們查看一下 Dataframe 的結(jié)構(gòu)：
Python:
tick_df.show(10)
輸出如下所示：
Plain Text:
+-------------------+-------+------+
| ts | price|symbol|
+-------------------+-------+------+
|2013-02-08 00:00:00| 45.08| A|
|2013-02-08 00:00:00| 14.75| AAL|
|2013-02-08 00:00:00| 78.9| AAP|
|2013-02-08 00:00:00|67.8542| AAPL|
|2013-02-08 00:00:00| 36.25| ABBV|
|2013-02-08 00:00:00| 46.89| ABC|
|2013-02-08 00:00:00| 34.41| ABT|
|2013-02-08 00:00:00| 73.31| ACN|
|2013-02-08 00:00:00| 39.12| ADBE|
|2013-02-08 00:00:00| 45.7| ADI|
+-------------------+-------+------+
only showing top 10 rows
現(xiàn)在準(zhǔn)備將 Dataframe 寫到 SingleStore。 在下一個(gè)代碼單元中，可以添加以下內(nèi)容：
Python:
%run ./Setup
在Setup筆記本中，需要確保已為 SingleStore 托管服務(wù)集群添加了服務(wù)器地址和密碼。
在下一代碼單元中，我們將為 SingleStore Spark 連接器設(shè)置一些參數(shù)，如下所示：
Python:
spark.conf.set("spark.datasource.singlestore.ddlEndpoint", cluster)
spark.conf.set("spark.datasource.singlestore.user", "admin")
spark.conf.set("spark.datasource.singlestore.password", password)
spark.conf.set("spark.datasource.singlestore.disablePushdown", "false")
最后，準(zhǔn)備使用 Spark Connector將 Dataframe 寫入 SingleStore
Python:
(tick_df.write
.format("singlestore")
.option("loadDataCompression", "LZ4")
.mode("ignore")
.save("timeseries_db.tick"))
這會(huì)將 Dataframe 寫入timeseries_db數(shù)據(jù)庫中的tick表?？梢詮?SingleStore 檢查該表是否已成功填充。
示例查詢
現(xiàn)在我們已經(jīng)構(gòu)建了系統(tǒng)，可以運(yùn)行一些查詢了。 SingleStore 支持一系列處理時(shí)間序列數(shù)據(jù)的有用函數(shù)。我們來看一些例子。
平均值
以下查詢說明了如何計(jì)算表中全部時(shí)間序列值的簡(jiǎn)單平均值：
SQL:
SELECT symbol, AVG(price)
FROM tick
GROUP BY symbol
ORDER BY symbol;
輸出應(yīng)該是：
Plain Text:
+--------+---------------+
| symbol | AVG(price) |
+--------+---------------+
| A | 49.20202542 |
| AAL | 38.39325226 |
| AAP | 132.43346307 |
| AAPL | 109.06669849 |
| ABBV | 60.86444003 |
... ...
時(shí)間分段
時(shí)間分段可以按固定時(shí)間間隔對(duì)不同時(shí)間序列的數(shù)據(jù)進(jìn)行聚合和分組。SingleStore 支持幾種函數(shù)：
- FIRST：與最小時(shí)間戳關(guān)聯(lián)的值。此文檔包含其他詳細(xì)信息和示例。
- LAST：與最大時(shí)間戳關(guān)聯(lián)的值。此文檔包含其他詳細(xì)信息和示例。
- TIME_BUCKET：將時(shí)間標(biāo)準(zhǔn)化為最近的存儲(chǔ)段的開始時(shí)間。此文檔包含其他詳細(xì)信息和示例。
例如，可以使用TIME_BUCKET查詢以五天為時(shí)間間隔進(jìn)行分組的平均時(shí)間序列值，如下所示：
SQL:
SELECT symbol, TIME_BUCKET("5d", ts), AVG(price)
FROM tick
WHERE symbol = "AAPL"
GROUP BY 1, 2
ORDER BY 1, 2;
輸出應(yīng)該是：
Plain Text:
+--------+-----------------------+--------------+
| symbol | TIME_BUCKET("5d", ts) | AVG(price) |
+--------+-----------------------+--------------+
| AAPL | 2013-02-08 00:00:00.0 | 67.75280000 |
| AAPL | 2013-02-13 00:00:00.0 | 66.36943333 |
| AAPL | 2013-02-18 00:00:00.0 | 64.48960000 |
| AAPL | 2013-02-23 00:00:00.0 | 63.63516667 |
| AAPL | 2013-02-28 00:00:00.0 | 61.51996667 |
... ... ...
還可以結(jié)合這些函數(shù)來創(chuàng)建燭臺(tái)圖，顯示股票隨時(shí)間的最高價(jià)、最低價(jià)、開盤價(jià)和收盤價(jià)，以五天為一個(gè)窗口單位，如下所示：
SQL:
SELECT TIME_BUCKET("5d") AS ts,
symbol,
MIN(price) AS low,
MAX(price) AS high,
FIRST(price) AS open,
LAST(price) AS close
FROM tick
WHERE symbol = "AAPL"
GROUP BY 2, 1
ORDER BY 2, 1;
輸出應(yīng)該是：
Plain Text:
+------------+--------+----------+----------+----------+----------+
| ts | symbol | low | high | open | close |
+------------+--------+----------+----------+----------+----------+
| 2013-02-08 | AAPL | 66.8428 | 68.5614 | 67.8542 | 66.8428 |
| 2013-02-13 | AAPL | 65.7371 | 66.7156 | 66.7156 | 65.7371 |
| 2013-02-18 | AAPL | 63.7228 | 65.7128 | 65.7128 | 64.4014 |
| 2013-02-23 | AAPL | 63.2571 | 64.1385 | 63.2571 | 63.5099 |
| 2013-02-28 | AAPL | 60.0071 | 63.0571 | 63.0571 | 60.0071 |
... ... ... ... ... ...
平滑
可以使用AVG對(duì)窗口進(jìn)行聚合來平滑時(shí)間序列數(shù)據(jù)。下面是一個(gè)示例，查看價(jià)格和過去三個(gè)分時(shí)價(jià)格的移動(dòng)均線：
SQL:
SELECT symbol, ts, price, AVG(price)
OVER (ORDER BY ts ROWS BETWEEN 3 PRECEDING AND CURRENT ROW) AS smoothed_price
FROM tick
WHERE symbol = "AAPL";
輸出應(yīng)該是：
Plain Text:
+--------+-----------------------+----------+----------------+
| symbol | ts | price | smoothed_price |
+--------+-----------------------+----------+----------------+
| AAPL | 2013-02-08 00:00:00.0 | 67.8542 | 67.85420000 |
| AAPL | 2013-02-11 00:00:00.0 | 68.5614 | 68.20780000 |
| AAPL | 2013-02-12 00:00:00.0 | 66.8428 | 67.75280000 |
| AAPL | 2013-02-13 00:00:00.0 | 66.7156 | 67.49350000 |
| AAPL | 2013-02-14 00:00:00.0 | 66.6556 | 67.19385000 |
... ... ... ...
截至
查找截至某個(gè)時(shí)間點(diǎn)的當(dāng)前表行也是常見的時(shí)間序列需求。這可以用ORDER BY和LIMIT輕松實(shí)現(xiàn)。下面是一個(gè)例子：
SQL:
SELECT *
FROM tick
WHERE ts <= "2021-10-11 00:00:00"
AND symbol = "AAPL"
ORDER BY ts DESC
LIMIT 1;
輸出應(yīng)該是：
Plain Text:
+-----------------------+--------+----------+
| ts | symbol | price |
+-----------------------+--------+----------+
| 2018-02-07 00:00:00.0 | AAPL | 159.5400 |
+-----------------------+--------+----------+
插值
時(shí)間序列數(shù)據(jù)可能存在缺值。我們可以插入缺失的點(diǎn)。 SingleStore文檔提供了一個(gè)示例存儲(chǔ)過程，可在處理tick數(shù)據(jù)時(shí)用于此目的。
加分：Streamlit 可視化
之前提到過燭臺(tái)圖，如果能以圖形而不是表格的形式看到這些圖表就太好了，而使用Streamlit可以輕松做到這一點(diǎn)。上一篇文章展示了我們可以輕松地將 Streamlit 連接到 SingleStore。
安裝所需軟件
我們需要安裝以下軟件包：
Plain Text:
streamlit
pandas
plotly
Pymysql
這些可以在GitHub上的requirements.txt文件中找到。運(yùn)行文件如下：
Shell:
pip install -r requirements.txt
示例應(yīng)用程序
以下是 streamlit_app.py 的完整代碼清單：
Python:
# streamlit_app.py
import streamlit as st
import pandas as pd
import plotly.graph_objects as go
import pymysql
# Initialize connection.
def init_connection():
return pymysql.connect(**st.secrets["singlestore"])
conn = init_connection()
symbol = st.sidebar.text_input("Symbol", value = "AAPL", max_chars = None, key = None, type = "default")
num_days = st.sidebar.slider("Number of days", 2, 30, 5)
# Perform query.
data = pd.read_sql("""
SELECT TIME_BUCKET(%s) AS day,
symbol,
MIN(price) AS low,
MAX(price) AS high,
FIRST(price) AS open,
LAST(price) AS close
FROM tick
WHERE symbol = %s
GROUP BY 2, 1
ORDER BY 2, 1;
""", conn, params = (str(num_days) + "d", symbol.upper()))
st.subheader(symbol.upper())
fig = go.Figure(data = [go.Candlestick(
x = data["day"],
open = data["open"],
high = data["high"],
low = data["low"],
close = data["close"],
name = symbol,
)])
fig.update_xaxes(type = "category")
fig.update_layout(height = 700)
st.plotly_chart(fig, use_container_width = True)
st.write(data)
創(chuàng)建機(jī)密文件
本地 Streamlit 應(yīng)用程序會(huì)從應(yīng)用程序的根目錄讀取機(jī)密文件 .streamlit/secrets.toml。需要按如下方式創(chuàng)建這個(gè)文件：
Plain Text:
# .streamlit/secrets.toml
[singlestore]
host = ""
port = 3306
database = "timeseries_db"
user = "admin"
password = ""
主機(jī)和密碼的應(yīng)替換為在創(chuàng)建集群時(shí)從 SingleStore 托管服務(wù)獲取的相應(yīng)值。
運(yùn)行代碼
可按如下方式運(yùn)行 Streamlit 應(yīng)用程序：
Shell:
streamlit run streamlit_app.py
在Web 瀏覽器中的輸出應(yīng)如圖 2 所示。
在網(wǎng)頁上，可以在文本框中輸入一個(gè)新的股票代碼，并使用滑塊來更改TIME_BUCKET的天數(shù)。隨意嘗試代碼以滿足您的需求。
總結(jié)
本文展示了 SingleStore 是處理時(shí)間序列數(shù)據(jù)的有效解決方案。利用 SQL 和內(nèi)置函數(shù)的強(qiáng)大功能，我們可以實(shí)現(xiàn)很多目標(biāo)。通過添加的FIRST、LAST和TIME_BUCKET，SingleStore 擴(kuò)展了對(duì)時(shí)間序列的支持。
致謝
感謝約翰·皮克福德(John Pickford) 博士對(duì)恰當(dāng)?shù)臅r(shí)間序列數(shù)據(jù)集的建議和指導(dǎo)。
此外還要感謝 Part-Time Larry制作的精彩視頻Streamlit--Building Financial DashBoards with Python 以及GitHub代碼，啟發(fā)了本文的 Streamlit 可視化部分。
譯者介紹
楊曉娟，社區(qū)編輯，西安電子科技大學(xué)計(jì)算機(jī)專業(yè)碩士研究生，資深研發(fā)工程師，信息系統(tǒng)項(xiàng)目管理師，擁有近20年Java開發(fā)經(jīng)驗(yàn)。分別在NEC、甲骨文、英方從事數(shù)據(jù)存儲(chǔ)、Oracle數(shù)據(jù)庫的數(shù)據(jù)遷移以及同構(gòu)/異構(gòu)數(shù)據(jù)庫復(fù)制等研發(fā)工作，尤其在數(shù)據(jù)庫、數(shù)據(jù)編碼等方面有深入鉆研和了解。
原文標(biāo)題：??Using SingleStore as a Time Series Database??，作者：Akmal Chaudhri
網(wǎng)站名稱：使用SingleStore作為時(shí)間序列數(shù)據(jù)庫
分享地址：http://fisionsoft.com.cn/article/cdssgec.html
