To be clear, I'm not evangelist of the language XD, some folks may have strong opinions against it. This article is just purely based on my 2 years' usage of the Rust programming language and its ecosystem. And I want to share my first hand knowledge, using my 5 years' experience in data engineering, to tell you the great things and the bad things Rust has to offer in this field.

1. Start with Python 🐍

About 5 years ago, when I was tired of being a python script kiddie, I was very lucky to be hired by a startup data company, where I learned everything about this industry -- batching, streaming, lakehouse, warehouse, extraction, transformation, ingestion... The company has different clients, big and small. Some has to deal with billions of records in their DBMS, some needs to do complex analytical computation on their private servers. And I was on the team responsible for our ETL toolkits, which we developed and integrated as a user friendly low-code platform. It was built around Pandas and Apache Spark, Pandas is for exploration and previewing, PySpark for batch processing, a very common data handling architecture. So naturally for me, I started working on it very quickly with my Monty Python knowledge.

Not so long after, however, we had some hard time with this architecture. I could spin up a chart/sheet very quickly using Pandas, but when the sample dataset became larger than usual sometimes, the OOM killer kicked in very timely. Classic pandas experience, and we had to increase the memory limit temporarily for it. Sometimes the data is too skewed, making one spark worker node crash repeatedly, dragging the whole spark cluster into a death loop. So we had to redo the partitioning, and the process is kinda slow and painful. You have to introduce some trial and error when the dataset is too large, doing any reasonable insights on it is very time and resource consuming. The story goes on and on...

2. Meet Rust 🦀

So, Pandas is smart, but it eats up exponential memory with bigger datasets. Spark can do the heavy lifting, but it's slow and not so flexible. Is there any tool smart/flexible enough, not so resource hungry, and can even do some huge data processing? We quickly eyed on the then newly born Polars project, which is mainly developed in Rust. It offers a similar experience with Pandas, and also provides a python package, so the most common things you do in Pandas can be done in Polars, just faster and using less memory, which is very convenient. However, Polars was not very feature complete then, many complex transformations was missing from it and it was not very extensive to develop with easily. And it does not feature a complete SQL engine, which was also needed by us.

Here comes Apache DataFusion then. It boasts a truly extensive query engine, also written in Rust, provides equal or better performance compared to Polars, supports hugely flexible data sources and data transformations. So we quickly did some experiments with it and was very satisfied. At that time, not many DBMSs was officially supported by it, we shamelessly dig from the Polars dependencies and found the connector-x crate, which is a high performance Rust connector to all kinds of DBMSs (BTW I made several PRs and were merged to it afterwards), and with several adapter traits and structs, we were able to connect DataFusion to all these MySQL, Postgres, Oracle, SQL Server... you name it, which was mind boggling and insanely productive. The Rust ecosystem was that vibrant and abundant and it's still thriving!

3. The Good Time

After several testing deployments replacing Spark, we were very satisfied with our DataFusion based query engine. It generally outperformed Spark not only in resource consumption, but also in speed. Even with much smaller datasets, it still managed to beat Spark due to node scheduling overhead in Spark, making realtime query execution possible. Clients and on-site engineers were also kinda surprised by this large improvement. So we decided to also publish it as an additional OLAP service into out ETL toolkits platform. Then things went even more smoothly (or rusty? lol). Thanks to the highly extensive nature of DataFusion and the fantastic tooling ecosystem of Rust, we managed to develop several SDK packages in mere weeks for our newly developed query engine, supporting Python and Go, making our other existing services blew their old versions out of the water as well, which blew our minds once again.

My first year with Rust went fly without me even realizing it. We had been busy constantly improving our query engine, building more supporting tools, online services, adding supports for more DBs, files and storage systems, doing all kinds of customizations for clients, and so on. Two full-time golang developers in our team were also converted to Rust (Rust is not cult, I promise). Overall, the Rust ecosystem is amazing, it's lively and rapidly evolving. You can find packages for all kinds of common functionalities on the public crates.io repository. There is an official dedicated docs.rs site for documents. The Rust book and the Rust async book are extremely helpful even to complete newbie programmers. Also, don't forget rust-by-example, the rust playground, and a very nice tour of rust if you are impressed by the tour of golang, and so on. There are countless Rust tutorials on the Internet, different types of learners can choose the best ones for themselves. Moreover, there're several online forums if you need extra help, like the r/rust subreddit, or the official rust users' forum. Many have the impression that Rust communities are toxic, hostile, or hard to work with. I can't say for everyone, but on our team, we generally find people in Rust communities are very eager to help. There is no "deep" knowledge kept away from new comers, hardly any "RTFM" humiliating comments which are quite common on the Stack Overflow sites, and I think are rude and unproductive. And to be fair, I personally find /r/rust much more helpful than SO, on /r/rust people can rant about all sorts of things Rust, help requests, advice seeks, new creation showoffs...serious or casual, they are not restricted to the rigid Question then Answer paradigm, people just exchange their knowledge freely, no superior or inferior. Though they can't avoid some issues from reddit's side, like information is hard to archive and search, formatting is messy, etc., but I think it's good enough for a free community for the time being.

4. The Bad Time

About another year went by, our team landed more clients than ever. But some little things we thought irrelevant went messy fast. First, we found the CI became unbearably slow while test builds on multiple branches were waiting in queue. It turned out to be a caching problem of the cargo building architecture, combined with the complex CI environment, there were no sound solution to this. We had to seek to dark magic to mitigate (I also wrote a post about this in Chinese). One day our devops colleague came up and ask what we were doing on the CI machine, our jobs consumed most of the CPU time, complaints from other teams were made from time to time, because their jobs kept crashing everyday. Yeah, you probably already knew it, cargo build is quite CPU intensive, and by default it eats up all the cores. We had to set limits to the builder containers, but the building time dragged long and long again. To reduce the wait for any changes to be landed, we even tried to broke our monolithic project into a modular, microservice design gradually, but quickly found out that the maintenance cost is punishing. You may ask, why not throw more machines to it? Truth is, we were still a startup, more machines calls for more maintenance efforts. And the CI cluster we had was by no means weak, around 256 cores, 1TB of ram in total, and a JBOD array god knows how many terabytes of disks it packed within. We were desperate to try to find ways to cut down the wait for compiling.

5. How We Improved

Here are some strategies we tried and had applied to the dependencies in cargo.toml. The core idea of it is being very careful about which crate and which version to pull in, in order to shrink the dependency tree as best as we can. a) Remove any unnecessary, or "left-pad" like dependency, re-implement in our own code base if possible. And now we know that the cargo build directory is just as heavy as node_modules.😅 b) Hand pick versions of the dependencies. Usually, there are overlapping among dependencies, dependencies of dependencies and so on, but just with different versions. Choose wisely on crates with a huge dependency tree, so that the result dependency tree could be as low and narrow as possible. c) If method b) doesn't work very well for some particular crate, mirror it on our own git server, re-export and/or modify some of its dependency. Sometimes this method might need us mirroring several other crates in a row, so we only use this if absolutely necessary.

Generally, these strategies helped us cut down about 50% of the dependency tree, and test build pipeline took us around 30 min with a limit of 16 CPU cores at that time. But it was still to wasteful compared to our non CGO golang projects, which generally take only about one or two minutes. And we had to find other ways. One thing we first thought about was the feature flag. Most of the time when we did the test builds, the relavant code paths were only limited to several specific packages within the code base. After we gated several lesser used packages behind feature flags, the time for one test build came down to around 20 mins. That was a nice but not that significant improvement.

Another thing we changed is the cargo build profile, see the Cargo Document on this topic. We were using release profile for the test builds before, because some test routes on non optimized debug build profile were too slow. We played with several options then decided to tune down the opt-level and lto options, along with the usage of mold the multithreaded modern linker, the build time was brought down to about 10 mins or less.

6. To be Continued

At this point, 10 mins for us is tolerable for now. But the rabbithole is way more deep, many more aspects we could still improve upon. And there are more rusty things I wanna share with you. More posts coming in...

Golang 自带了内容相当丰富的 io 库，对 io 编程友好。最近碰到个需要发送文件的 REST api，文件内容需 base64 编码为文本，同时还要文件的 md5sum，再附加一系列其他信息，最后用 json 文本发送。最无脑的做法是将整个文件读入缓冲区，依次计算 md5sum，再把文件编码为 base64 文本，最后拼接所需的 json 文本完成发送。当然这样文件一大铁定崩内存💩，显然有更合适的做法……

稍稍好一点的做法是从头到尾读取一次文件计算 md5sum，再 seek 回文件头部再从头到尾读一次完成编码，最后拼接 json 文本并发送。这种做法省掉了一次文件读取的缓冲，但还有 base64 编码的缓冲没有去掉，因为最后拼接 json 的时候仍然需要完整的内容，仍会占用大块内存，文件一大还是会崩。此处的源码以及运行效果如下，其中 echo.go 为模拟的 API 服务器，会原样返回客户端上传的数据，upload.go 为本程序：

package main

import (
	"bytes"
	"crypto/md5"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
)

func main() {
	f, err := os.Open("./testFile.bin")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()

    // 计算 md5sum
	hasher := md5.New()
	_, err = io.Copy(hasher, f)
	if err != nil {
		log.Fatal(err)
	}
	md5sum := fmt.Sprintf("%x", hasher.Sum(nil))

    // 文件回退到头部
	_, err = f.Seek(0, io.SeekStart)
	if err != nil {
		log.Fatal(err)
	}

    // base64 编码文件，编码后的文本存入 encoded
	encoded := bytes.NewBuffer(nil)
	encoder := base64.NewEncoder(base64.StdEncoding, encoded)
	buf := make([]byte, 1024)
	for {
		n, err := f.Read(buf)
		if err != nil && err != io.EOF {
			log.Fatal(err)
		}
		if n == 0 {
			break
		}
		encoder.Write(buf[:n])
	}
	encoder.Close()

    // 拼接 json
	jsonBytes, err := json.Marshal(map[string]string{
		"md5":  md5sum,
		"file": encoded.String(),
	})
	if err != nil {
		log.Fatal(err)
	}

    // 发送到服务 API
	httpClient := &http.Client{}
	res, err := httpClient.Post(
		"http://127.0.0.1:8081/",
		"application/json",
		bytes.NewBuffer(jsonBytes),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer res.Body.Close()

    // 读取服务响应
	bodyBytes, err := io.ReadAll(res.Body)
	if err != nil {
		log.Fatal(err)
	}

	log.Printf(
		"client: response Status: %s\n"+
			"body length: %d\n"+
			"body preview: %s",
		res.Status, len(bodyBytes), string(bodyBytes[:100]),
	)
}

下面是本篇要介绍的方法，无需额外的大缓冲区，通过结合使用 go 自带的 io.Pipe、io.TeeReader、io.MultiReader、io.Copy 等，用一次文件读取完成了所有操作。其中 io.TeeReader 在读取文件 Reader 的同时，把内容抄写到另一个 md5 hasher 里来计算 md5sum 值；接着再抄写到 base64 编码器中，同时由 io.Pipe 完成写转读（而不是用缓冲区）得到一个读取编码后内容的 reader，最后用 io.MultiReader 把各个 reader 串接出单一 reader chained，http.Client.Post 用它一次便能读取到完整内容的 json，无需任何缓冲区直接发送。大体流程示意图和详细代码如下。

func upload2() {
	f, err := os.Open("./testFile.bin")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()

	hasher := md5.New()
	copiedReader := io.TeeReader(f, hasher)
	hashBuffer := bytes.NewBuffer(nil)

	pr, pw := io.Pipe()
	encoder := base64.NewEncoder(base64.StdEncoding, pw)
	encoderDoneChan := make(chan interface{})
	go func() {
		_, err := io.Copy(encoder, copiedReader)
		if err != nil {
			pw.CloseWithError(err)
			log.Fatal(err)
		} else {
			pw.Close()
		}
		encoder.Close()
		hashBuffer.WriteString(fmt.Sprintf("%x", hasher.Sum(nil)))
		close(encoderDoneChan)
	}()

	jsonHead := bytes.NewReader([]byte(`{"fileContent":"`))
	jsonHead2 := bytes.NewReader([]byte(`","md5":"`))
	jsonTail := bytes.NewReader([]byte(`"}`))
	chained := io.MultiReader(jsonHead, pr, jsonHead2, hashBuffer, jsonTail)

	httpClient := &http.Client{}
	res, err := httpClient.Post(
		"http://127.0.0.1:8081/",
		"application/json",
		chained,
	)
	if err != nil {
		log.Fatal(err)
	}
	defer res.Body.Close()
} 

点击展开/折叠查看示意图

First things first, to achieve that, we need a proxy where we can set up port forwarding on it, then we modify and compile qbittorrent ourselves. A VPN can work, but that requires UPnP/PMP support of the VPN, which many lacks in. And also, use with caution, if you do not want to expose your IP address accidentally, always use a VPN.

Recently, I came across a situation where I need to do some seeding, but the network I was on requires a proxy to gain access to the Internet. Naively, I set up the proxy settings in qbittorrent, quickly found out that it did not work at all.

After some long nights searching and digging, it turned out that the library qbittorrent (along with several other FOSS torrenting software) depends on, libtorrent, on some recent versions, completely disables port listening while Use proxy for peer connections is enabled, to avoid network traffic leakage, rendering it impossible to do seeding at all while using a proxy. But the authors never make this clear, and I suspect that qbittorrent authors haven't even realized this behavior. The setting page on qbittorrent doesn't reflect the change in port listening section at all when you enable proxy for peers.

So here come the easier parts. We find the corresponding conditional code, modify, patch, then recompile libtorrent. No changes in qbittorrent are needed, only libtorrent. Check out my patch file listen-when-proxy-on.patch based on v2.0.7 of libtorrent at end of this article.

Since I'm on archlinux (and yes, BTW, I'm using arch :D), patching software is relatively convenient. Go grab the PKGBUILD script from arch offical package page https://archlinux.org/packages/extra/x86_64/libtorrent-rasterbar/ (source files link), add our patch into prepare() section like this:

prepare() {
    makedir -p build
    cd "$pkgname-$pkgver"
    patch --forward --strip=1 --input="${srcdir}/listen-when-proxy-on.patch"
}

and do makepkg -Acs to compile the package. After it's done, sudo pacman -U libtorrent... to do a force install. Then we need to do some config stuff.

• Fire up qbittorrent and set a fixed listening port, say 7890;
• Then set up port forwarding on the proxy server we're gonna use. I'm using FRP for port forwarding on my proxy server, so I set up a tcp reverse tunnel with local_port as 7890 on my computer;
• Go to qbittorrent advanced settings page, enable Allow multiple connections from the same IP address;
• Finally, set proxy server for qbittorrent, and enable proxy for peer connections. Do not turn on UPnP/NAT-PMP. But if you know what you're doing, the patch ensures that they can be turned on along with proxy.

Now we can add a torrent and see, it should start listening for incoming connections from the reverse tunnel we just set on the proxy server. And we can also see all the incoming peers shown as localhost or 127.0.0.1. No worries, the bittorrent protocol doesn't care about peers' ip addresses, from everyone else's perspective, you're just running qbittorrent on 7890 on the proxy server, everything works fine.

Disclaimer: qBittorrent and libtorrent are FOSS, the patch shared here is also free from restrictions. Please comply with your local regulations, use with caution and at your own risks.

listen-when-proxy-on.patch:

diff --git a/src/session_impl.cpp b/src/session_impl.cpp
index e25434926..8f38a3344 100644
--- a/src/session_impl.cpp
+++ b/src/session_impl.cpp
@@ -2016,7 +2016,7 @@ namespace {
 
 		// if we don't proxy peer connections, don't apply the special logic for
 		// proxies
-		if (m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
+		if (false && m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
 			&& m_settings.get_bool(settings_pack::proxy_peer_connections))
 		{
 			// we will be able to accept incoming connections over UDP. so use
@@ -2774,7 +2774,7 @@ namespace {
 		// directly.
 		// This path is only for accepting incoming TCP sockets. The udp_socket
 		// class also restricts incoming packets based on proxy settings.
-		if (m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
+		if (false && m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
 			&& m_settings.get_bool(settings_pack::proxy_peer_connections))
 			return;
 
@@ -5473,7 +5473,7 @@ namespace {
 			// connections. Not even uTP connections via the port we know about.
 			// The DHT may use the implied port to make it work, but the port we
 			// announce here has no relevance for that.
-			if (sock->flags & listen_socket_t::proxy)
+			if (false && sock->flags & listen_socket_t::proxy)
 				return 0;
 
 			if (!(sock->flags & listen_socket_t::accept_incoming))
@@ -5513,7 +5513,7 @@ namespace {
 			return std::uint16_t(sock->tcp_external_port());
 		}
 
-		if (m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
+		if (false && m_settings.get_int(settings_pack::proxy_type) != settings_pack::none
 			&& m_settings.get_bool(settings_pack::proxy_peer_connections))
 			return 0;
 
@@ -5611,7 +5611,7 @@ namespace {
 
 		if (!s->natpmp_mapper
 			&& !(s->flags & listen_socket_t::local_network)
-			&& !(s->flags & listen_socket_t::proxy))
+			&& !(s->flags & listen_socket_t::proxy && false))
 		{
 			// the natpmp constructor may fail and call the callbacks
 			// into the session_impl.
@@ -6833,8 +6833,9 @@ namespace {
 		// there's no point in starting the UPnP mapper for a network that isn't
 		// connected to the internet. The whole point is to forward ports through
 		// the gateway
+        // Enable UPnP anyway, even when on proxy, we can deal with this ourselves.
 		if ((s->flags & listen_socket_t::local_network)
-			|| (s->flags & listen_socket_t::proxy))
+			|| (s->flags & listen_socket_t::proxy && false))
 			return;
 
 		if (!s->upnp_mapper)

最近做某演示，期间几个 ssh 操作把对方开发给惊到了，我才发现 OpenSSH 有些奇技淫巧并不是所有开发都知道的，特来水上一篇。

几年前翻 OpenSSH 文档时无意中发现，隧道转发可以在现有连接上动态设置，而无需重开或者另开连接。看了看现在中文互联网上关于 ssh 隧道的一些文章，都没见到有关动态设置的说明，都默认开启隧道需另建或重建新的 ssh 连接。其实不然，OpenSSH 客户端/服务端支持在保持连接的同时增删隧道。某些特殊场景下，在技术防护不到位的环境里用好这点可以做出隧道嵌套，端口封锁逃逸等等各种骚操作🤐。

具体操作：已经连上 ssh 时，按一下 Enter，再按 Shift+`（也就是 ~），最后按 Shift+c 调出 OpenSSH 客户端的内置命令行，输入 help 帮助，输出所有隧道操作的命令说明如下。

接下来该怎么开启各类隧道不必我赘述了，再次调出 ssh 客户端的内置命令行，输入相关参数回车即可。

还有一些其他操作，搜索 "ssh escaping sequences" 即可。例如 Enter，~，. 强行断开 ssh 连接，Enter，~，Ctrl+z 把 ssh 客户端置于后台运行等等。

前段时间遇到一个略有挑战性的数据需求，要求高速处理巨量的标志位数据。我费了些折腾功夫在 Spark 里用 SQL 的二进制操作给实现了个，但总觉得应该有更优雅的实现，有一种动作变形的感觉，虽然目前还挺堪用。

下边这是大致需求。有个大宽表 T（大约不超过 5k 列），对 T 的每条数据做某种计算可以得到它们的一些状态，状态的种类是固定的，比如这里的数据只能有 A、B、C、D 四种状态中的一种或多种。现在要求保存 T 每一列每一行所有数据的状态，以便能被后续步骤继续使用，在保证数据完整性的同时，除了不能让结果数据过分膨胀，还要减少在后续步骤里的额外 io 开销/计算开销，避免影响正常业务……大概这就是既要马儿跑又要马儿不吃草😂？

一般对简单不是非常大量的数据，线上逻辑是干 json 就完了，因为数据量不大的缘故，几乎不会有什么序列化反序列化压缩解压缩的开销，并且 json 本身是肉眼可读数据，对所有人都方便。

大宽表里再用 json 显然是不可行的，业务内少说就有上百列、至少上千万行的状态要处理，可能除了代码比较简单好写以外，光序列化反序列化的性能损耗就够难受了。对这种时间/空间两难的境地，咱第一反应是自定义底层数据结构。然而对 Spark 这样固定封装了数据类型的框架来讲，改源码去定义新类型显然不是最优解。所以转念想有没有可能用已有数据类型，做一个自定义的抽象类型呢？翻了翻 Spark 数据类型相关的文档，所有基础原始类型都有固定长度、固定范围，嗯嗯很好，这些类型定义是统一的。再去翻了翻 SQL 层文档，看看有哪些可以操作原始类型的 SQL 函数，诶不错，左右位移，与或非异或之类也齐活了……

0. 定义数据结构

由于状态种类是固定的，对每条数据而言所有可能出现状态的情形也是固定了，下面就自然想到可以用类似 bitset 的方法去做状态的存储，利用率高，解析读取几乎没有计算量，相对来讲对时间空间都友好。还是 ABCD 四种状态为例，我们用 1 表示某状态存在，0 表示某状态不存在。比如说 0000 表示 ABCD 四种状态都没有，1000 表示只有 A，1001 表示有 A 和 D，0110 表示有 BC，1111 表示 ABCD 都有……

1. 定义存储方法

现在可以用 4 个比特位表示一个数据的状态，那么问题又来了，怎么才能把这些状态存入原有表中呢？Spark DataFrame 里显然没有四个比特位（半个字节）的基础类型，那我们拿个常用的整型 IntegerType 来用好啦，32 位的数据，存四个比特绰绰有余。下边拿 pyspark 举例说明：

>>> from pyspark.sql import (
...     SparkSession,
...     functions as F,
...     types as T,
... )

# 创建 session
>>> spark = SparkSession.builder.appName('app').getOrCreate()

# 读入原表文件
>>> dataframe = spark.read.parquet('/path/to/file')

# 假定只有三列
>>> dataframe.columns
['column1', 'column2', 'column3']

# 假设计算第一列四种状态的 sql 如下，
# 每条 sql 返回 0 或 1，返回 1 表示某状态存在，0 则不存在
# column2, column3 其他两列以此类推
>>> column1_A_sql = F.when(..., 1).otherwise(0)
>>> column1_B_sql = F.when(..., 1).otherwise(0)
>>> column1_C_sql = F.when(..., 1).otherwise(0)
>>> column1_D_sql = F.when(..., 1).otherwise(0)

# 四个状态位合并到一个 IntegerType 整型中
>>> column1_states = F.lit(0) \
...     .bitwiseOR(F.shiftLeft(columns1_A_sql, 3)) \
...     .bitwiseOR(F.shiftLeft(columns1_B_sql, 2)) \
...     .bitwiseOR(F.shiftLeft(columns1_C_sql, 1)) \
...     .bitwiseOR(F.shiftLeft(columns1_D_sql, 0)) \
...     .astype(T.IntegerType())

# 类推得到另两列的完整状态 column2_states, column3_states，
# 执行以上 sql 得到包含三列完整状态的 df `states`
>>> states = dataframe.select(
...     column1_states.alias('column1_states')
...     column2_states.alias('column2_states')
...     column3_states.alias('column3_states')
... )
>>> states.show()
+--------------+--------------+--------------+
|column1_states|column2_states|column3_states|
+--------------+--------------+--------------+
|             1|             4|             7|
|             2|             5|             8|
|             3|             6|             9|
|           ...|           ...|           ...|
+--------------+--------------+--------------+

2. 定义读取/解析方法

比特存储的标识位数据，读取起来也很方便。

# 每种标识位所在的顺序
>>> state_index = {'A': 3, 'B': 2, 'C': 1, 'D': 0}

# 定义从一列中读取某种标识状态的函数
>>> def get_state_x(states_dataframe, state_column, state_type):
...     # 获取状态对应的顺序位，状态名不对时抛出错误
...     index = state_index.get(state_type)
...     if index is None:
...         raise Exception(f'Unknown state type of {state_type}')
...
...     # 过滤掉其他状态位
...     masked_state = F.col(state_column).bitwiseAND(1 << index)
...
...     # 判断剩下的状态位上是否还有 1，如果只剩 0，表明当前状态不存在，否则为存在
...     result = F.when(masked_state == F.lit(0), F.lit(False)) \
...         .otherwise(F.lit(True))
...
...     # 对状态 df 执行 sql
...     return states_dataframe.select(result.alias(f'{state_type}_exists'))
...

# 继续拿上边得到的 states 举例，解析 column1 中是否有 D 状态
>>> get_state_x(states, 'column1_states', 'D').show()
+--------+
|D_exists|
+--------+
|    true|
|   false|
|    true|
|     ...|
+--------+

到这里就完成了完整流程。

3. 优化改进

上边的实现方式有一个明显可改进之处 —— 使用 32 位的 IntegerType 存储 4 位的数据，只存了一列的状态，剩余 28 位浪费掉了，是不是可以在一个 32 位整型中最多塞入 8 列的状态，直接提升 7 倍的利用率呢？这思路是可行的，我们甚至还可以把整表的状态全塞入一列中

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
from more_itertools import ichunked

spark = SparkSession.builder.appName('app').getOrCreate()

# 假设这次原表有 1000 列数据的状态要处理, column1, column2, column3, ...
table = spark.read.parquet('/path/to/table')

# 假设计算一列中某状态的 sql 函数如下，同之前一样，仍为返回 0 或 1 的 sql
def column_state_A(column)：
    ...
    
def column_state_B(column):
    ...
    
def column_state_C(column):
    ...

def column_state_D(column):
    ...
    
# 获取全表状态的函数，一列有 ABCD 四种状态，八列一组，共 32 位，存入一个 32 位整型
# 最后把所有整数做成一个数组，存入一个数组列中
def get_full_states(table):
    # 创建一个整数数组列，用来存入全表的所有状态
    all_states_column = F.array().astype(T.ArrayType(T.IntegerType()))
    
    # 每 8 列取一组，每组生成一个整型存入组内所有状态
    for chunk in ichunked(enumerate(table.columns), 8):
        state_int = F.lit(0).astype(T.IntegerType())
        
        # 遍历组内每一列
        for column_index, column in chunk:
            # 计算本列的每种状态
            state_A = column_state_A(column)
            state_B = column_state_B(column)
            state_C = column_state_C(column)
            state_D = column_state_D(column)
            
            # column_index 为本列在原表中序号，
            # 计算本列在本组内的序号，因为是 8 列一组，所以可得组内序号如下
            ingroup_index = column_index % 8
            
            # 然后可得本列每个状态在组内的具体序号
            a_index = ingroup_index * 4 + 3
            b_index = ingroup_index * 4 + 2
            c_index = ingroup_index * 4 + 1
            d_index = ingroup_index * 4
            
            # 写入本列的全部状态到组内
            for state in [
                F.shiftLeft(state_A, a_index),
                F.shiftLeft(state_B, b_index),
                F.shiftLeft(state_C, c_index),
                F.shiftLeft(state_D, d_index)
            ]:
                state_int = state_int.bitwiseOR(state)
                
        # 一组的状态合并到全表
        all_states_column = F.concat(
            all_states_column, 
            F.array(state_int.astype(T.IntegerType()))
        )
        
    # 返回生成全表状态列的完整 sql
    return all_states_column

# 生成所需 sql，并执行得到全表的状态列，放入表中，列名为 all_states
all_states_sql = get_full_states(table)
table = table.withColumn('all_states', all_states_sql)

预览看下最后生成的全表状态列，大概长这个样

>>> table.select(F.col('all_states')).show()
+-----------+
| all_states|
+-----------+
|[10 10 ...]|
|  [0 0 ...]|
|  [0 1 ...]|
+-----------+

核心步骤和改进前是一致的，只是多了缩减使用空间的步骤：将多列合并为一组，一组写入到同一个整数，最后将所有组的整数合并到一个数组列中。

接着读取状态的解析函数如下：

# 接上，生成的全表状态列为 all_states
# 同样，四种状态的相对顺序序号
state_index = {'A': 3, 'B': 2, 'C': 1, 'D': 0}

def get_state_x_from_column(table, all_states, column, state):
    """
    计算某列中包含某状态的情况
    
    :param table:        带有全表状态列的整表
    :param all_states:   全表状态列的列名
    :param column:       需要关注的列名
    :param state:        需要关注的状态名
    """
    columns = table.columns
    columns.remove(all_states)
    if column not in columns:
        raise Exception(f'列名 {column} 不正确，表中没有此列')
        
    if state not in state_index:
        raise Exception(f'状态名 {state} 不正确，没有此状态')
        
    # 使用本列在全表中的序号，计算本列所在组的组序号
    group_index = columns.index(column) // 8
    # 计算本列在组内的序号
    column_ingroup_index = columns.index(column) % 8
    # 计算所需状态在组内的序号
    state_ingroup_index = state_index.get(state) + 4 * column_ingroup_index
    
    # 根据以上序号过滤出所需状态的标识位
    state_masked = F.col(all_states).getItem(group_index).bitwiseAND(1 << state_ingroup_index)
    
    # 根据标识位返回 true/false 表示本列包含此状态的情况，返回完整 sql
    return F.when(state_masked == 0, False).otherwise(True)

我们看下使用效果

>>> sql = get_state_x_from_column(table, 'all_states', 'column1', 'A')
>>> table.select(sql.alias('A_in_column1')).show()
+------------+
|A_in_column1|
+------------+
|        true|
|       false|
|       false|
+------------+

Perfecto!

4. 其他

目前这段代码已经上线，实测生成全表状态占用的计算时间为任务总用时的 0.01% ~ 0.1%，解析某列状态占用的计算时间稳定在个位数毫秒，几乎可忽略不计。空间占用的增加量同所需状态量、数据表行列总数呈简单的线性关系，同表数据本身关系不大。

最后效果还不错，不过说到底还得归功于 Spark 这样的神级开源软件，默默在背后提供了如此完整又强大的功能，让我等小白能混口饭吃😂

之前 GitHub 上很多人用的 Netflix 解锁 Docker 容器，是用一个专门的 sniproxy 程序“反代” HTTPS 流量。sniproxy 的好处是配置够简单灵活，坏处是如果已经有 HTTPS 服务在运行，那 443 端口被占了是没法用 sniproxy 的。本站用 Nginx 反代后端，而 Nginx 早已经可以通过 stream 变相实现 sniproxy 的功能，于是就有了本文的做法，可以让解锁服务和普通 HTTPS 网站同时运行。

先说一下，Netflix 目前已经逐步放松 IP 地理位置限制了，原来完全不能用的地址现在至少可以看自制剧，也有很多很棒的剧集哟！第三方内容里大部分还是需要代理解锁，比如 AMC 的 The Walking Dead，Better Call Saul 等等。下面就讲讲 DNS 解锁 Netflix 是怎么一回事。

0. 准备

首先，我们需要一个能看 Netflix 完整内容的 IP，就是没有被它标记限制的那种… 嗯，既然我已经有可以看的 IP 了还要解锁干啥呀？别急，配好这个解锁可以给电视电话电冰箱电磁炉用或者给分享朋友等等用，而不需要安装额外的软件配置额外的代理，跟普通的翻墙代理不一样，这里只设置一个 DNS 就可以。题外话，据说目前私有 IPv6 基本没有限制，有大佬自己注册了 ASN 买了段 v6 地址亲测可用，这个超纲了，不在本文讨论范围内😂。

1. Nginx

SSH 登上没有被 Netflix 屏蔽的服务器，设置一下 Nginx。参考 Nginx 有关文档。如果已经装有 Nginx 的话，nginx -V 执行看下可用的 flag 里有没有打开编译选项
--with-stream_ssl_preread_module，如果没有的话，ubuntu 可以 apt install 安装 nginx-extras 替换掉原来的版本，安装完全体 Nginx，其他发行版应该也有类似包，实在不行再自行配置选项重编译吧。

Nginx 修改配置，nginx.conf 中找到 stream {} 区域，如果没有的话可以自己手动添加。里面加一个 server 用来“反代” HTTPS 数据流，再加一组正则映射规则通过 SNI 过滤域名，像这样，只允许 Netflix 域名的 SNI：

stream {
    # 加一组 map，根据规则过滤 $ssl_preread_server_name （也就是请求的 SNI）
    # 将过滤过的 SNI 存入自定义变量 $filtered_sni_name，
    # 如果没有匹配到任何规则，则默认为一个 443 端口没有服务的本地回环。
    map $ssl_preread_server_name $filtered_sni_name {
        ~^(.*|)netflix\.com$ $ssl_preread_server_name;
        ~^(.*|)netflix\.net$ $ssl_preread_server_name;
        ~^(.*|)nflximg\.com$ $ssl_preread_server_name;
        ~^(.*|)nflximg\.net$ $ssl_preread_server_name;
        ~^(.*|)nflxvideo\.net$ $ssl_preread_server_name;
        ~^(.*|)nflxext\.com$ $ssl_preread_server_name;
        ~^(.*|)nflxso\.net$ $ssl_preread_server_name;
        ~^(.*|)fast\.com$ $ssl_preread_server_name;
        default "127.255.255.255";
    }

    server {
        resolver 1.1.1.1;
        listen 443;
        ssl_preread on;
        # 只代理经过 SNI 过滤的流量，
        # 其他流量都被丢入不存在的地址
        proxy_pass $filtered_sni_name:443;
    }
}

保存完了重载 Nginx 即可。这里加 map 来过滤流量的意义在于防止服务器被滥用成任意网站的反代。

2. Dnsmasq

如果到这里你已经明白是怎么一回事的话，大概不需要 DNS 你也知道怎么弄了── hosts 文件写入上边这些域名，指向那个服务器的 IP 就行。当然为了方便使用我们继续，假定上边没被 Netflix 封锁的 IP 是 a.b.c.d，现在任意找一个延迟较低的公网服务器，或者继续用 a.b.c.d 也可以。假定这次的 IP 叫 1.2.3.4。在 1.2.3.4 上的 Dnsmasq 增加配置如下：

address=/netflix.com/a.b.c.d
address=/netflix.net/a.b.c.d
address=/nflxvideo.net/a.b.c.d
address=/nflxext.com/a.b.c.d
address=/nflximg.com/a.b.c.d
address=/nflximg.net/a.b.c.d
address=/nflxso.net/a.b.c.d
address=/fast.com/a.b.c.d

保存后重启 Dnsmasq 即可。

3. 使用

现在告诉你的朋友，让他把网络设置的 DNS 改成 1.2.3.4 就可以使用 Netflix 咯，电视机 PlayStation Xbox 啥的也是改个 DNS 就能用。只要使用者可以正常连接 a.b.c.d 和 1.2.3.4 即可。如果使用者连接不到这俩地址的话🤔那得配合中转代理更复杂的玩法，留到以后再写吧。