首先用pip安裝下面導的包

1.操作Mysql

 1 import logging
 2 import pymysql
 3 
 4 
 5 def connect_test_db():
 6     return pymysql.connect(host='10.169.30.190',
 7                            port=3306,
 8                            user='root',
 9                            password='123456',
10                            database='
 
test',
11                            charset='UTF8')
12 
13 
14 def select_sql():
15     connect = connect_test_db()  # 連線物件
16     cursor = connect.cursor()  # cursor遊標
17 
18     try:
19         sql_str = "SELECT * FROM test.Student"
20         cursor.execute(sql_str)  # 執行語句
21         realist = cursor.fetchall()  #
 
 拿到執行結果
22 
23         for row in realist:
24             print(row)
25 
26     except:
27         logging.exception('select operation error')
28         raise
29     finally:  # 關流
30         cursor.close()
31         connect.close()
32 
33 
34 def insert_sql():
35     connect = connect_test_db()  #
 
 連線物件
36     cursor = connect.cursor()  # cursor遊標
37     sql_str = "INSERT INTO test.Student VALUES('09','張三','1996-03-11','男')"
38 
39     try:
40         connect.begin()  # 開啟事務
41         cursor.execute(sql_str)
42         connect.commit()  # 提交sql
43 
44         print('insert succeed')
45     except Exception as e:  # 如果出現異常執行下面語句
46         print(e)
47         connect.rollback()  # 回滾
48     finally:
49         cursor.close()
50         connect.close()
51 
52 
53 if __name__ == '__main__':
54     select_sql()

2.操作Redis

import redis

pool = redis.ConnectionPool(host='10.169.30.190',
                            port=6379,
                            db=1)
r = redis.Redis(connection_pool=pool)


# redis預設埠是6379
def select():
    r.set('name', 'Jack')
    r.set('age', 18, ex=2)  # ex - 過期時間（秒）
    r.set('name', 'Tom', nx=True)  # nx - 如果值為True則key存在不插入
    print(r.get('name'))


if __name__ == '__main__':
    select()

3.操作ClickHouse

注意：這裡的埠是TCP協議的9000埠不是http的8123埠

import pandas
from clickhouse_driver import Client

client = Client(host='10.169.30.190',
                port='9000',
                user='default',
                password='default',
                database='default')


def select():
    sql = client.execute("SELECT * FROM default.clickStream")
    result = pandas.DataFrame(sql)  # 執行語句並拿到返回結果
    print(result)


def insert():
    sql = client.execute("INSERT INTO default.clickStream VALUES ('customer1', '2021-10-02', 'add_to_cart', 'US', "
                         "568239 )")
    result = pandas.DataFrame(sql)
    print(result)


if __name__ == '__main__':
    select()