Shawn`s Blog

由于 Clickhouse-fdw 依赖高版本的 GNU tool chain 需要使用 GCC 7+ 来编译， 并且依赖的一些第三方库的版本也比较高， 原生 CentOS 的 repo 不能满足依赖，所以用选用 Conda 环境来解决依赖并 build 生成动态库。

先配置 clickhouse 表并导入数据

CREATE DATABASE test_database;

USE test_database;

CREATE TABLE tax_bills_nyc
(
    bbl Int64,
    owner_name String,
    address String,
    tax_class String,
    tax_rate String,
    emv Float64,
    tbea Float64,
    bav Float64,
    tba String,
    property_tax String,
    condonumber String,
    condo String,
    insertion_date DateTime MATERIALIZED now() 
)
ENGINE = MergeTree PARTITION BY tax_class ORDER BY (owner_name)

curl -X GET 'http://taxbills.nyc/tax_bills_june15_bbls.csv'  \
  | clickhouse-client --input_format_allow_errors_num=10 \
  --query="INSERT INTO test_database.tax_bills_nyc FORMAT CSV"

配置启动 postgresql,首先要初始化 postgresq

initdb pdatta

这个时候会以 ec2-user 用户创建一个 pdata 的目录，里面会有生成 postsql 的默认配置。 修改 pdata 目录下的两个文件 postgresql.conf 和 pg_hba.conf ， 第一个文件修改 listernling port 和 bind ip 改为 0.0.0.0, 第二个文件修改 trust 的 ip, 这样就可以从外部访问 postgresql.

pg_ctl start -D pdata -s -w -t 300

如果需要停止服务可以运行

pg_ctl stop -D pdata -s -m fast

创建 clickhouse-fdw extension

CREATE EXTENSION clickhousedb_fdw;

创建 SERVER, 这里需要指定 clickhouse 的地址为 host, 以及之前编译的 clickhouseodbc.so 地址作为 driver 的地址, dbname 也是 clickhouse 上的 =dbname=。

CREATE SERVER clickhouse_svr FOREIGN DATA WRAPPER clickhousedb_fdw OPTIONS(dbname 'test_db', driver '/home/ec2-user/workspaces/clickhouse-odbc/build/driver/libclickhouseodbc.so', host '172.31.37.147');

创建user mapping

CREATE USER MAPPING FOR CURRENT_USER SERVER clickhouse_svr;

创建外表, 并且表名需要跟 clickshouse 的表名一致。

CREATE FOREIGN TABLE tax_bills_nyc
(
     bbl int8,
     owner_name text,
     address text,
     tax_class text,
     tax_rate text,
     emv Float,
     tbea Float,
     bav Float,
     tba text,
     property_tax text,
     condonumber text,
     condo text,
     insertion_date Time
 ) SERVER clickhouse_svr;

查询数据

最近在做项目的时候遇到稍微复杂的一个场景,需要夸多组件数据处理。具体流程如下：

• 从 google bigquery 大叔据平台将计算和聚合过的每小时结果集存在 pandas 的 dataframe 中。
• pandas 的数据与 postgresql 数据库同步, 将结果集中有些字段为空的部分从 postgresq 中再回填给 pandas 的 dataframe 中。
• 合并之后的结果在更新入 postgresq, 同时写入 hive 表中便于后期用 presto 做长查询。

本来这套逻辑似乎更适用于 spark map reduce 来做，但是由于 bigquery 的计算分布在几十个 gcp project 里面。 spark 调用 bigquery 的话配置权限很麻烦，同时结果集每小时的大小也只在三万到十万级别。 计算任务是在 bigquery 中完成, 所以这时候调用 spark 似乎有点重了。所以决定使用 python pandas 配合别的大数据主件。

bigquer 每次会计算历史10几个小时的聚合结果集，按照每小时3-10万的量算下来，最多一次合并 postgresq 的量级在百万以内。 这个量级在合并的时候从 pgsql 拉数据下来，再在内存中用 pandas 做 join 明显不是明智的选择。 从 pgsql 中 select 百万记录下来 pandas join 完之后再上传 pqsql ，再 update 无疑增加了计算的复杂度。 所以我的选择是: 按小时将 pandas dataframe 上传到 pgsql 临时表， 直接在数据库中做 join 并 update pgsql 中的记录 。

因此我们需要实现的就是如何优雅的把 pandas dataframe 上传到 pgsql 的临时表，并提供接口操作临时表和已有的表做运算。

虽然 pandas 自己也带了数据库连接的接口，但是过于简单，不能满足复杂的处理过程。 但是又不想用很重的库如 sqlalchemy 等支持 ORM 的库，觉得还是对程序员不友好。而且要写很多手脚架来适配。

所以这里推荐我常用的数据库连接库 records

• 动态解析 pandas dataframe 对应的数据库临时表的 schema

  def infer_pd_schema(df): '''infer pandas dataframe schema ''' schema = [] type_map = { 'int64': 'int8', 'object': 'varchar(512)', 'float64': 'float8' } for col, tp in df.dtypes.items(): t = type_map[tp.name] item = f'''"{col}" {t}''' schema.append(item) return ', '.join(schema)

  如果要增加跟多复杂的类型作为 pandas dataframe 和 pgsql 的映射，可以往 type_map 里面添加对应的类型。 或者你也可以修改非 pgsql 的类型数据库对应的字段，以匹配你真实的需求。

• 上传 pandas dataframe 数据到数据库临时表

  def create_temp_table_with_df(conn, table, dataframe): '''conn is a pgsql connect transaction table is the name of the temp table and df is the pandas dataframe ''' schema = infer_pd_schema(dataframe) conn.query(f'drop table if exists {table}') query = f'''create TEMPORARY table if not exists {table} ({schema});''' logger.info(query) conn.query(query) # After testing, threads will not improve the uploading performance # here is a tricky way to upload a pandas dataframe to pgsql without "insert" query. # 600,000 records can be delivered within secs. with tempfile.NamedTemporaryFile() as tf: name = tf.name logger.info(f'creating temp data file {name}') dataframe.to_csv(name, header=True, index=False) query = f''' copy {table} from STDIN WITH (FORMAT CSV, HEADER TRUE) ''' rconn = conn._conn.connection cursor = rconn.cursor() with open(name, 'r') as f: cursor.copy_expert(query, f) rconn.commit() logger.info('upload data done')

  当 pandas 的数据量很大比如几百万条记录时， sql insert 操作的性能往往不会很好。即便是使用多线程操作，性能的提升也是不明显。 这里我采用了 pgsql 的 copy 命令， 很 tricky 的深入了 records 库的低层 connection 调用方法。 百万级的记录上传只需要秒级就可以完成。

• 对 pandas dataframe 的临时表与数据库中已经有的表做联合操作

  def query_pg_with_df(table, dataframe, sql): '''Temp upload pandas dataframe to pgsql for interaction with tables in pg. return a pandas dataframe ''' pgdb = records.Database( 'postgres://{host}/{db}?user={user}&password={pwd}'.format( host=PG_SQL_HOST, db=PG_SQL_NAME, user=PG_SQL_USER, pwd=PG_SQL_PASS) ) try: with pgdb.transaction() as conn: create_temp_table_with_df(conn, table, dataframe) logger.info(sql) res = conn.query(sql) return pd.DataFrame(res.as_dict()) except Exception as e: logger.error(e.msg) finally: pgdb.close()

  这里函数中的 table 变量是指后面 pandas dataframe 上传至 pgsql 之后的临时表名。变量 sql 就可以操作这个临时表和数据库里面其他表做正常的类似 join 的操作。 返回的结果是一个 pandas dataframe 。 这样就可以在内存中做后续进一步的处理了。

• 如果采用 sql insert 一条条纪录的话，性能不能符合要求，所以采用了 copy csv 的方式。
• 提供了 pandas dataframe 如何与 pgsql 里的表做交互比如 join 等操作。
• 可以根据实际采用的 sql 服务类型，修改此代码 porting 到 mysql 或者别的 sql service 。

• 首先你需要一个 Linux 的 VPS ，以及一个可以用的域名。 VPS 的操作系统可以选 Ubuntu 或者 centos 等主流的 Linux 系统。 同时删除原来系统里面的 sentmail 相关的 package 。
• 在你的域名供应商的平台上配置好你的域名解析。 一般情况下，需要配置3种域名解析类型， 分别是 A, MX, TXT 下面是我的配置, 我用的是阿里云的域名服务：

• 在 VPS 上安装好 postfix ， dovecot ， sasl-dev 等软件。 libsasl 是 postfix 和 dovecot 认证用的。你需要根据你的 VPS 操作系统的类型选择对应的包管理工具安装这些包。

  postfix-2.10.1-7.el7.x86_64
  dovecot-2.2.36-3.el7_7.1.x86_64
  dovecot-mysql-2.2.36-3.el7_7.1.x86_64
  cyrus-sasl-lib-2.1.26-23.el7.x86_64
  cyrus-sasl-plain-2.1.26-23.el7.x86_64
  cyrus-sasl-2.1.26-23.el7.x86_64
  

• 在 VPS 上创建一个用户，你的邮箱就是这个用户名，目前没有尝试用 LDAP 服务和 postfix 服务做集成。
• 最好还能有一个可以有一个常用的别的邮箱如 gmail 等作为 relay sever, 这是因为 私有的邮件服务器发送出去的邮件会被认为是垃圾邮件 。 据说是可以配置 domain 的 TXT 类型的记录可以解决这个问题，但是我没尝试成功。

在测试 smtp 和 imap 协议的时候， 对 encrypted session only 使用的端口默认是需要用 openssl 来访问的。 而非 encrypted 可以直接使用 telnet 来测试。

• 测试 SMTP 服务 调用 openssl 访问服务器的 465 端口：

  openssl s_client -connect mail.example.com:465

  返回 ssl 协商的结果

  New, TLSv1.2, Cipher is ECDHE-RSA-AES256-GCM-SHA384
  Server public key is 4096 bit
  Secure Renegotiation IS supported
  Compression: NONE
  Expansion: NONE
  No ALPN negotiated
  SSL-Session:
      Protocol  : TLSv1.2
      Cipher    : ECDHE-RSA-AES256-GCM-SHA384
      Session-ID: 4D22C02B32E6B83A7380CE7C69A69FEE3985CE7B653EB423CDCA29940
      Session-ID-ctx:
      Master-Key: 9E10F874F9052D5EE47DE164A3DC95A008A3829D6E7CBD071473D9313ADC3FD9DE9DE
      PSK identity: None
      PSK identity hint: None
      SRP username: None
      TLS session ticket lifetime hint: 3600 (seconds)
      TLS session ticket:
      0000 - 07 ad 7b 9e c4 12 81 3f-13 06 ed c9 72 3a 08 e7   ..{....?....r:..
      0010 - 0e f2 72 84 70 18 47 17-94 b2 05 94 1a a7 6a 6c   ..r.p.G.......jl
      0020 - c5 37 04 ee f7 c0 36 0f-29 44 67 a9 cb f6 91 14   .7....6.)Dg.....
      0030 - 72 b6 21 45 fa 82 3b 8e-51 76 5b 4a 2e 6c 26 2a   r.!E..;.Qv[J.l&*
      0040 - 15 0c a3 3d 2c ed de ee-41 04 26 0c 89 93 c3 4f   ...=,...A.&....O
      0050 - e9 84 a0 46 68 73 b7 f4-94 3f 46 a9 af 37 a4 7f   ...Fhs...?F..7..
      0060 - 2f 3c 73 bc 43 8b 75 ac-5f 33 10 60 f6 d4 ca 74   /<s.C.u._3.`...t
      0070 - a6 60 49 8f bc 7e be 73-1e 47 c5 6d 50 21 95 53   .`I..~.s.G.mP!.S
      0080 - 9b 78 82 d6 0d d5 32 20-eb 94 50 bc a6 b1 6f fe   .x....2 ..P...o.
      0090 - 03 ee 80 4d 09 35 47 14-e4 5d aa d2 18 1f c7 ee   ...M.5G..]......
      00a0 - ca 11 a6 c1 3f d3 66 42-df 3b 4d 66 0e 7e 95 89   ....?.fB.;Mf.~..
      00b0 - 62 d0 69 76 3b fd 74 7a-d7 86 d5 6a d0 23 f8 b3   b.iv;.tz...j.#..
  
      Start Time: 1584262970
      Timeout   : 7200 (sec)
      Verify return code: 18 (self signed certificate)
      Extended master secret: no
  ---
  220 mail.example.com ESMTP Postfix
  

  输入 auth login plain 的命令, 后面跟上你的用户名密码的 base64 编码。 其中 account, password 用你真实的账号替换。

  echo -ne "\0account\0password" | base64

  220 machineheart.tech ESMTP Postfix
  AUTH LOGIN PLAIN AGFjY291bnQAcGFzc3dvcmQ=
  501 5.5.4 Syntax: AUTH mechanism
  AUTH LOGIN
  334 VXNlcm5hbWU6
  

  看到类似的返回即表示验证成功。后面即可以使用 smtp 的常用命令类似 mail from:, rcpt to: 去发送邮件了。

• 测试 IMAP 服务 与 smtp 类似，也是使用 openssl 来测试 993 端口。

  openssl s_client -connect mail.example.com:993

   *tag login account password*
   tag OK [CAPABILITY IMAP4rev1 LITERAL+ SASL-IR LOGIN-REFERRALS ID ENABLE IDLE SORT SORT=DISPLAY THREAD=REFERENCES 
   THREAD=REFS THREAD=ORDEREDSUBJECT MULTIAPPEND URL-PARTIAL CATENATE UNSELECT CHILDREN NAMESPACE UIDPLUS LIST-EXTENDED 
   I18NLEVEL=1 CONDSTORE QRESYNC ESEARCH ESORT SEARCHRES WITHIN CONTEXT=SEARCH LIST-STATUS BINARY MOVE 
   SNIPPET=FUZZY SPECIAL-USE] Logged in
   *tag list "" "*"*
   * LIST (\HasNoChildren \Sent) "." Sent
   * LIST (\HasNoChildren \Drafts) "." Drafts
   * LIST (\HasNoChildren) "." "Deleted Items"
   * LIST (\HasNoChildren) "." INBOX
  tag OK List completed (0.001 + 0.000 secs).