You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
54 KiB
54 KiB
None
<html>
<head>
</head>
</html>
Benchmark: ClickHouse Vs. InfluxDB Vs. Postgresql Vs. Parquet¶
How to use:¶
- Rename the file "properties-model.ini" to "properties.ini"
- Fill with your own credentials
The proposal of this work is to compare the speed in read/writing a midle level of data ( a dataset with 9 columns and 50.000 lines) to four diferent databases:
- ClickHouse
- InfluxDB
- Postgresql
- Parquet (in a S3 Minio Storage)
ToDo: - DuckDB with Polars
- MongoDB
- Kdb+
Deve-se relevar:
é uma "cold-storage" ou "frezze-storage"?
influxdb: alta leitura e possui a vantagem da indexaçõa para vizualização de dados em gráficos.
notas:
- comparar tamanho do csv com parquet
Imports¶
In [18]:
import configparser
from datetime import datetime
import duckdb
import influxdb_client
import pandas as pd
# import pymongo
from clickhouse_driver import Client
from dotenv import load_dotenv
from minio import Minio
from pymongo import MongoClient
from pytz import timezone
from sqlalchemy import create_engine
load_dotenv()
Out[18]:
In [1]:
# Variables
dbname = "EURUSDtest"
In [5]:
arq = configparser.RawConfigParser()
arq.read("properties.ini")
ClickHouseUser = arq.get("CLICKHOUSE", "user")
ClickHouseKey = arq.get("CLICKHOUSE", "key")
ClickHouseUrl = arq.get("CLICKHOUSE", "url")
InfluxDBUser = arq.get("INFLUXDB", "user")
InfluxDBKey = arq.get("INFLUXDB", "key")
InfluxDBUrl = arq.get("INFLUXDB", "url")
InfluxDBBucket = arq.get("INFLUXDB", "bucket")
PostgresqlUser = arq.get("POSTGRESQL", "user")
PostgresqlKey = arq.get("POSTGRESQL", "key")
PostgresqlUrl = arq.get("POSTGRESQL", "url")
PostgresqlDB = arq.get("POSTGRESQL", "database")
S3MinioUser = arq.get("S3MINIO", "user")
S3MinioKey = arq.get("S3MINIO", "key")
S3MinioUrl = arq.get("S3MINIO", "url")
S3MinioRegion = arq.get("S3MINIO", "region")
MongoUser = arq.get("MONGODB", "user")
MongoKey = arq.get("MONGODB", "key")
MongoUrl = arq.get("MONGODB", "url")
In [ ]:
%%time
# Load Dataset
df = pd.read_csv("out.csv", index_col=0)
In [ ]:
# df.head()
In [ ]:
df["from"] = pd.to_datetime(df["from"], unit="s")
df["to"] = pd.to_datetime(df["to"], unit="s")
# Optional use when not transoformed yet
# Transform Datetime
Funçoes¶
-> Class
In [ ]:
def timestamp2dataHora(x, timezone_="America/Sao_Paulo"):
d = datetime.fromtimestamp(x, tz=timezone(timezone_))
return d
ClickHouse¶
In [ ]:
# !! O client oficial usa um driver http, nesse exemplo vamos usar a biblioteca
# de terceirtos clickhouse_driver recomendada, por sua vez que usa tcp.
client = Client(
host=ClickHouseUrl,
user=ClickHouseUser,
password=ClickHouseKey,
settings={"use_numpy": True},
)
In [ ]:
# Create Tables in ClickHouse
# !! ALTERAR TIPOS !!
# ENGINE: 'Memory' desaparece quando server é reiniciado
client.execute(
"CREATE TABLE IF NOT EXISTS {} (id UInt32,"
"from DateTime, at UInt64, to DateTime, open Float64,"
"close Float64, min Float64, max Float64, volume UInt32)"
"ENGINE MergeTree ORDER BY to".format(dbname)
)
In [ ]:
%%time
# Write dataframe to db
client.insert_dataframe("INSERT INTO {} VALUES".format(dbname), df)
In [ ]:
%%time
client.query_dataframe("SELECT * FROM default.{}".format(dbname)) # LIMIT 10000
In [ ]:
# %%time
# df = pd.DataFrame(client.query_dataframe("SELECT * FROM default.{}".format(dbname)))
InfluxDB¶
In [ ]:
client = influxdb_client.InfluxDBClient(
url=InfluxDBUrl, token=InfluxDBKey, org=InfluxDBUser
)
In [ ]:
# Read data from CSV without index and parse 'TimeStamp' as date.
df = pd.read_csv("out.csv", sep=",", index_col=False, parse_dates=["from"])
# Set 'TimeStamp' field as index of dataframe # test another indexs
df.set_index("from", inplace=True)
In [ ]:
df.head()
In [ ]:
%%time
# gravando... demorou... mas deu certo
with client.write_api() as writer:
writer.write(
bucket=InfluxDBBucket,
record=df,
data_frame_measurement_name="id",
data_frame_tag_columns=["volume"],
)
In [ ]:
# data
# |> pivot(
# rowKey:["_time"],
# columnKey: ["_field"],
# valueColumn: "_value"
# )
In [ ]:
# Read
Postgresql¶
In [ ]:
# Connect / Create Tables
engine = create_engine(
"postgresql+psycopg2://{}:{}@{}:5432/{}".format(
PostgresqlUser, PostgresqlKey, PostgresqlUrl, PostgresqlDB
)
)
In [ ]:
# Drop old table and create new empty table
df.head(0).to_sql("comparedbs", engine, if_exists="replace", index=False)
In [ ]:
%%time
# Write
conn = engine.raw_connection()
cur = conn.cursor()
output = io.StringIO()
df.to_csv(output, sep="\t", header=False, index=False)
output.seek(0)
contents = output.getvalue()
cur.copy_from(output, "comparedbs") # , null="") # null values become ''
conn.commit()
cur.close()
conn.close()
In [ ]:
# Read
S3 Parquet¶
In [ ]:
# fazer sem funçao para ver se melhora
# verifique se esta no ssd os arquivos da pasta git
def main():
client = Minio(
S3MinioUrl,
secure=False,
region=S3MinioRegion,
access_key="MatMPA7NyHltz7DQ",
secret_key="SO1IG5iBPSjNPZanYUaHCLcoSbjphLCP",
)
# Make bucket if not exist.
found = client.bucket_exists("data")
if not found:
client.make_bucket("data")
else:
print("Bucket 'data' already exists")
# Upload
client.fput_object(
"data",
"data.parquet",
"data/data.parquet",
)
# print(
# "'data/data.parquet' is successfully uploaded as "
# "object 'data.parquet' to bucket 'data'."
# )
In [ ]:
%%time
df.to_parquet("data/data.parquet")
if __name__ == "__main__":
try:
main()
except S3Error as exc:
print("error occurred.", exc)
In [ ]:
pq = pd.read_parquet("data/data.parquet", engine="pyarrow")
pq.head()
MongoDB¶
In [10]:
# Load csv dataset
data = pd.read_csv("out.csv")
In [36]:
# Connect to MongoDB
client = MongoClient(
# "mongodb://192.168.1.133:27017"
"mongodb://{}:{}@{}/EURUSDtest?retryWrites=true&w=majority".format(
MongoUser, MongoKey, MongoUrl
),
authSource="admin",
)
In [37]:
db = client["EUROUSDtest"]
collection = db["finance"]
# data.reset_index(inplace=True)
data_dict = data.to_dict("records")
In [38]:
%%time
# Insert collection
collection.insert_many(data_dict)
Out[38]:
In [ ]:
# read
DuckDB¶
In [39]:
cursor = duckdb.connect()
print(cursor.execute("SELECT 42").fetchall())
In [45]:
%%time
conn = duckdb.connect()
data = pd.read_csv("out.csv")
conn.register("EURUSDtest", data)
Out[45]:
In [47]:
display(conn.execute("SHOW TABLES").df())
In [46]:
%%time
df = conn.execute("SELECT * FROM EURUSDtest").df()
df
Out[46]:
Kdb+¶
In [66]:
import numpy as np
np.bool = np.bool_
from qpython import qconnection
In [49]:
# read csv
data = pd.read_csv("out.csv")
In [50]:
# open connection
q = qconnection.QConnection(host="localhost", port=5001)
q.open()
In [51]:
# send df to kd+ in memory bank
%%time
q.sendSync("{t::x}", data)
In [52]:
# write to on disk table
q.sendSync("`:/home/sandman/q/tab1 set t")
Out[52]:
In [53]:
%%time
# read from on disk table
df2 = q.sendSync("tab2: get `:/home/sandman/q/tab1")
In [54]:
# print(df2)
In [55]:
%%time
# load to variable df2
df2 = q.sendSync("tab2")
In [58]:
# df2(type)
In [67]:
%%time
# converto to dataframe
df = pd.DataFrame(q("t", pandas=True))
df.head()
In [19]:
%%time
# select
df3 = q.sendSync("select from t")
In [ ]:
q.close()