SQLでNULL同士を等しいとみなして効率的に比較を行う方法

今回はSQLの小ネタです。

初心者がミスりがちな話なのですが、SQLでは通常 NULLとNULLは等しいとは見做されません。

例えば、`select null = null ` など実行すると、Trueではなくnull が返ってきます。

しかし、場合によってはnull同士は等しいものとして判定したいことがあり、その場合は何かしら一工夫する必要があります。両方nullという場合だけTrue返せばいいということもなく、当然値が入っているなら値が入ってるもの同士は通常の比較処理を行い、0や空文字も含めてnull以外の値とnullは違うものとみなし、その上でnull同士は等しいという判定をやるケースですね。

null以外の値が全部0以上の数値であることがわかっているなら coalesce でnullを-1に変換してから比較するとか、文字列方の列で、かつ値が入っている部分にnullって文字列がないことが確認できているなら null っていう文字列で埋めて比較するといった手段が取れます。

しかし、この列に絶対無いと言い切れる値が存在しない場合、補完して比較する方法は使えません。こういった場合に、スマートにnullを考慮した比較を行える方法をMySQLとSnowflakeの両環境について紹介します。

MySQLの場合

MySQLの場合、 <=> という演算子がサポートされています。これは、「NULLセーフイコール演算子」といいます。

これを使うと、 `selct null <=> null` がTrueになります。

ドキュメント: MySQL :: MySQL 8.0 Reference Manual :: 14.4.2 Comparison Functions and Operators

Snowflakeの場合

Snowflakeの場合、上記の<=>演算子はサポートされていませんが、`is disticnt from` という演算子が実装されています。

ドキュメント: IS [ NOT ] DISTINCT FROM | Snowflake Documentation

ちょっと長いので、<=>のほうが便利だよなぁとは思うのですが、標準SQLに準拠した書き方はこちらの方です。(最初、Snwoflake専用の方言かと勘違いしていました)

そこそこの頻度で使う機会がある構文だと思うので、頭の片隅にでも置いといてください。

Pythonで非負値行列分解

これもずっと前に記事にしたような気がしていたのですが、最近ちょっと仕事で使おうと思って自分のブログで検索したら書いてなかったことがわかったのでこの機会に記事にします。

行列を複数の行列の積に分解する方法は複数ありますが、その中でも非負値行列分解というsy方があります。

これは、元の行列の全ての要素が0以上(0を許すので正値ではなく非負値といいます)の場合に使える分解方法です。一般的には低rankな二つの行列で、それぞれの行列の全ての要素も0以上の行列の積に分解します。

数式で言うと、$V \approx W \times H$ ですね。 $V$が$M\times N$行列だとした場合、分解後のrankを$K$とすると、$W$は$M\times K$行列、$H$は$K\times N$行列になります。

ここで$V$は元の行列(データ行列)で、行数はサンプル数、列数は特徴数です。
$W$は基底行列になり各行は元データの異なる要素の「基底」や「パターン」を示します。
そして、$H$は係数行列で、各列は基底行列の要素がどの程度元のデータに寄与しているかを示します。

この分解によって、元のデータの背後に潜在する低次元のパターンや構造を捉えることができます。

分解後の要素が全て非負なので、分解結果を加法的に扱えるのが利点です。負の値が混ざってるとある値が大きかった時にそれに掛け算される係数が正なのか負なのか考慮して解釈しないといけないですがここが絶対0以上と保証されていると評価されやすいですね。

また、次元削減やデータ量の削減にも有宇高です。この用途で使われるため、$K$は小さな値が採用されやすいです。

この非負値行列分解はが画像処理とか音声解析、推薦システムの中で活用されていますね。

さて、scikit-learnを使って実際にやってみましょう。乱数で生成した行列でやってみますね。

ドキュメントはこちらです。
参考: NMF — scikit-learn 1.5.2 documentation

import numpy as np
from sklearn.decomposition import NMF

# サンプルデータ生成
# 乱数で5x4の非負行列を作成
np.random.seed(0)
V = np.random.randint(0, 6, size=(5, 4))

# NMFを適用、ランク(分解する際の次元)を2に設定
model = NMF(n_components=2, init='random', random_state=0)
W = model.fit_transform(V)  # 基底行列W
H = model.components_       # 係数行列H

# 分解結果を表示
print("元の行列 V:")
print(V)
"""
元の行列 V:
[[4 5 0 3]
 [3 3 1 3]
 [5 2 4 0]
 [0 4 2 1]
 [0 1 5 1]] 
 """

print("\n基底行列 W:")
print(W)
"""
基底行列 W:
[[1.97084006 0.        ]
 [1.35899323 0.33537722]
 [0.90516396 1.61014842]
 [0.76977917 0.6670942 ]
 [0.         1.7698575 ]]
"""

print("\n係数行列 H:")
print(H)
"""
係数行列 H:
[[2.09286365 2.49350723 0.         1.50627845]
 [0.63319723 0.41601049 2.69948881 0.        ]]
"""

# 元の行列の近似値を計算
V_approx = np.dot(W, H)

print("\n近似された行列 V_approx:")
print(V_approx)
"""
近似された行列 V_approx:
[[4.12469952 4.91430393 0.         2.96863391]
 [3.05654746 3.52817989 0.90534704 2.04702222]
 [2.91392627 2.92687151 4.34657765 1.36342897]
 [2.03344504 2.19696811 1.80081332 1.15950177]
 [1.12066887 0.73627928 4.7777105  0.        ]]
"""

見ての通りでちょっとクセがありますね。
基底行列の方はtransformで元のデータを変換して取得し、係数行列の方がcomponents_に入っています。

さて、近似した行列ですが、元の行列が純粋にただの乱数で生成されたもので、通常のデータであれば背景にあるはずの隠れた構造とかを一切持たないものだった割に結構近い値で近似できてるのではないでしょうか。

久しぶりに使うと、どっちの行列がどっちだっけとか、転地必要だったっけ、とか色々迷うのですが慣れれば手軽に扱えるので機会があれば試してみてください。

Pythonでマルチプロセス処理

前回の記事がマルチスレッドだったので今回はマルチプロセスを紹介します。

Pythonにおけるマルチプロセスの1番のメリットはGILの制約を回避できることでしょうね。

ただ、先に書いておきますが、この記事で書いている方法はJupyter notebookのセルに直接書くと正常に動作せずエラーになることがあります。.pyファイルを作成してそこに記入して使うようにしましょう。

マルチプロセスを実装するには、最近はconcurrent.futuresのProcessPoolExecutorを使います。
参考: concurrent.futures — 並列タスク実行 — Python 3.12.6 ドキュメント

ドキュメントのサンプルコードを参考に動かしてみましょう!
例として取り上げられているのは素数判定ですね。Pythonで処理が完結するのですが、GIL制約のためマルチスレッドだと高速化の恩恵が受けられないものです。

from concurrent.futures import ProcessPoolExecutor
import math


PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419
    ]

def is_prime(n):
    print(f"整数 {n} を素数判定します")
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d は素数か: %s' % (number, prime))

if __name__ == '__main__':
    main()

# 以下実行結果
"""
整数 112272535095293 を素数判定します
整数 112582705942171 を素数判定します
整数 112272535095293 を素数判定します
整数 115280095190773 を素数判定します
整数 115797848077099 を素数判定します
整数 1099726899285419 を素数判定します
112272535095293 は素数か: True
112582705942171 は素数か: True
112272535095293 は素数か: True
115280095190773 は素数か: True
115797848077099 は素数か: True
1099726899285419 は素数か: False
"""

最初にそれぞれの値の素数判定が始まってる旨のメッセージが出てその後に結果が順番に出てきたので、並行して処理されているのが確認できました。

is_prime(n)が並行して実行している処理です。

ProcessPoolExecutor() でエクゼキューターを作成して、今回は submit()ではなく、mapで適用していますね。map()には第一引数で並列実行したい関数を渡し、次の引数でその関数に渡す引数のリストを渡します。

submit と map はどちらもProcessPoolExecutor や ThreadPoolExecutor の継承元の抽象クラスのExecutor に実装されているメソッドなので、実はマルチプロセスとマルチスレッドのどちらでも両方使うことができます。お好みの方で書いたらよさそうです。

細かい挙動は異なっていて、前回のsubmit()ではas_completed()を使って終わった順番に処理を取り出していましたが、map()を使う場合は、処理自体は並列して同時に行われて順不同で完了しますが、結果の取り出しは渡した引数の順番になります。

Pythonでマルチスレッド処理

とっくの昔に、threadingを使ったマルチスレッド処理について記事を書いていたつもりだったのに、まだ書いてないことに気づきました。(そして、マルチプロセスの処理についてもまだ書いてませんでした。)

それでは気づいたこのタイミングで記事にしようと思ったのですが、改めてドキュメントを見てみると、concurrent.futures というより高レベルなモジュールがあるとのことでしたので、こちらを利用したマルチスレッド処理について紹介します。

先に言っておきますが、PythonにはGIL (Global Interpreter Lock) という制約があって、マルチスレッドにしたとしても、Pythonインタープリタは一度に1つのスレッドしか実行できません。なので、Pythonで完結するプログラムはマルチスレッドしても高速化の恩恵はありません。では、いつマルチスレッドは使うのかというと、Python外部のリソース(ストレージとかOSの処理とかWebアクセスとか)の待ち時間が発生する場合になります。

前置きが長くなってきましたが、実際に、concurrent.futuresを使ったマルチスレッドの並列処理のサンプルコードを紹介します。concurrent.futures.ThreadPoolExecutor というのを使います。
参考: concurrent.futures.ThreadPoolExecutor

5つのサイトへのアクセスを並列でやってみましょう。

import concurrent.futures
import requests
import time


# 取得するURLのリスト
URLS = [
    'http://www.example.com',
    'http://www.python.org',
    'http://www.openai.com',
    'http://www.wikipedia.org',
    'http://www.github.com'
]


# URLからコンテンツを取得する関数
def fetch_url(url):
    print(f"実行開始: {url}")
    response = requests.get(url)
    print(f"実行完了: {url}")
    return url, response.status_code, len(response.content)


# マルチスレッドでURLを並列取得する
start_time = time.time()

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # 各URLに対してfetch_url関数を並列実行
    futures = {executor.submit(fetch_url, url): url for url in URLS}

    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        try:
            url, status, content_length = future.result()
            print(f"URL: {url}, Status: {status}, length: {content_length}")
        except Exception as e:
            print(f"{url}でエラーが発生しました: {e}")

print(f"処理時間: {time.time() - start_time}秒")

# 以下結果
"""
実行開始: http://www.example.com
実行開始: http://www.python.org
実行開始: http://www.openai.com
実行開始: http://www.wikipedia.org
実行開始: http://www.github.com
実行完了: http://www.python.org
URL: http://www.python.org, Status: 200, length: 50928
実行完了: http://www.github.com
URL: http://www.github.com, Status: 200, length: 254186
実行完了: http://www.openai.com
URL: http://www.openai.com, Status: 403, length: 14186
実行完了: http://www.example.com
URL: http://www.example.com, Status: 200, length: 1256
実行完了: http://www.wikipedia.org
URL: http://www.wikipedia.org, Status: 200, length: 78458
処理時間: 0.49734020233154297秒
"""

ドキュメントのコードをもとにしていますが、fetch_url()メソッドの最初と最後にprit文を差し込んで5つのURLについて同時に処理が進んでいるのが分かるようにしました。開始と終了が異なる順番で結果がprintされていて、並列で動いてた感がありますね。

さて、上記コードの fetch_url() がマルチスレッドで実行されていた関数本体ですが、 肝心のThreadPoolExecutorはかなり使い方にクセがあります。

oncurrent.futures.ThreadPoolExecutor(max_workers=5) でエグゼキューターを作って、submit()や、as_completed()というメソッドを使っていますね。

submit() は実行キューへタスクを送信するメソッドです。

そして、もう一つ、oncurrent.futures.as_completed() というのを使っています。
こちらは、送信された非同期タスクが完了した順にFutureオブジェクトを返すジェネレータ関数です。これを使うことで、並列で動いていたメソッドが完了した順に、後続の処理を行うことができます。
上の例では、future.result() でメソッドの戻り値を受け取って、順次printしています。

使い所は慎重に選ばないと高速化等の効果は得られないですし、書き方にクセがあるので、慣れないと少々戸惑うのですが、ハードウェアアクセスの待ち時間が長い時や外部リソースへのアクセスを伴う処理の高速化では非常に役に立つものなので機会があったら使ってみてください。

Streamlitでアニメーション

今回はStreamlitでアニメーションを作成します。

といっても、やることは以前紹介したプレースホルダーの中身を順次更新し続けるだけ、という実装です。
参考: Streamlitのコンテナを使って動的にページを表示する

アニメーションさせるためには一つの枠を連続的に書き換えて画像を表示するので、st.empty() を使います。

とりあえず一個やってみましょう。画像の描写はmatplotlibを使ってみました。お試しなのでアニメーションの内容は線分をぐるぐる回すだけです。(両端を三角関数で実装します。)

import streamlit as st
import matplotlib.pyplot as plt
import numpy as np


# 描画エリアを設定
fig = plt.figure()
ax = fig.add_subplot(111)

ax.set_xlim(-1.2, 1.2)
ax.set_ylim(-1.2, 1.2)

# アニメを描写するプレースホルダーを作成
placeholder = st.empty()

# Streamlitのアニメーション表示
for i in range(100):
    ax.clear()
    ax.set_xlim(-1.2, 1.2)
    ax.set_ylim(-1.2, 1.2)
    ax.plot(
        [np.cos(i*0.1), -np.cos(i*0.1)],
        [np.sin(i*0.1), -np.sin(i*0.1)]
    )
    
    # プレースホルダーを更新
    placeholder.pyplot(fig)

これで線がぐるぐる回るアニメーションが表示できます。

あれ、time.sleep(0.01)とかウェイトを入れておかないとこのfor文が一瞬で終わってしまうんじゃないの?と思われるかもしれませんが、実験してみたところちょうど良い感じにアニメーションになりました。

どうもstreamlitの仕様として一枚一枚の画像の表示(pyplot)にウェイトがかかっているようです。

これは結構大きなメリットで、あまり表示時間とか気にせずにいい感じのアニメーションが作れます。

一方で、time.sleep(0.01) で0.01秒間隔の表示で1000フレーム使ってピッタリ10秒の動画を作ろう!みたいな調整は困難です。まぁ、これはstreamlitは動画作成を念頭に置いたものではないと思うので仕方ないですね。

ただし、デフォルトだと動作が早すぎるという場合はtime.sleep()を使ってウェイトを増やしましょう。