白い馬、走れ！「スーホの白い馬」に学ぶ、非同期処理とKafkaによるデータパイプライン構築

データ牧場の大レース！～同期処理の落とし穴～

広大な草原に、色とりどりの旗がはためいていた。殿様が主催する競馬大会だ。スーホと白い馬は、スタートラインに立っていた。

「白い馬、頼むぞ。勝ったら、お前には美味しい草を腹いっぱい食べさせてやる」

白い馬は、ひひんと鳴いて、スーホの言葉に答えた。まるで風のように、白い馬は他の馬を抜き去り、ゴールテープを切った。

「やった！白い馬、すごいぞ！」

しかし、殿様は約束を守らなかった。「この見事な馬は、わしのものだ！」と叫び、白い馬を連れ去ろうとした。

「殿様！約束が違う！白い馬は私の友達です！」

スーホの訴えもむなしく、白い馬は奪われてしまった。この出来事は、まるでプログラムがエラーで止まってしまうようだ。殿様は、システムのボトルネックのように、全てを自分のものにしようとして、全体の処理を止めてしまう。

def スーホの競馬(馬):
    結果 = 走る(馬)
    if 結果 == "一位":
        約束を守る(馬, "美味しい草")  # 殿様はこれを無視！同期処理が破綻！
    else:
        残念がる(スーホ)

def 走る(馬):
  if 馬 == "白い馬":
    return "一位"
  else:
    return "その他"

def 約束を守る(馬, 景品):
    if 殿様の気分 == "良い":  # ここで殿様の気分に依存しているのが問題！
        与える(馬, 景品)
    else:
        奪う(馬)  # エラー発生！処理が止まる

def 残念がる(スーホ):
    print("ああ、負けてしまった...")

def 与える(馬, 景品):
    print(f"{馬}に{景品}をあげます！")

def 奪う(馬):
    print(f"{馬}を奪います！")
    raise Exception("殿様が馬を奪った！システムエラー発生！") # 例外で処理中断を表現

殿様の気分 = "悪い" # 殿様の気分が悪いとエラー発生！
スーホの競馬("白い馬")

このコードのように、殿様の行動は予測できない。まるでシステムに予期せぬエラーが発生したかのようだ。非同期処理ならば、殿様の行動に関わらず、スーホは白い馬に草を与える処理を続けられたかもしれない。同期処理の落とし穴を、スーホは身をもって体験したのだった。

逃亡劇！～非同期処理の幕開け～

白い馬は、殿様の兵士たちに囲まれた。槍を構えた兵士たちが白い馬に迫る。

「白い馬！逃げろ！」スーホは叫んだ。

白い馬は、ひひんと高く鳴き、草原を駆け抜けた。矢が雨のように降り注ぐ。白い馬は、矢を避けながら、必死に走り続ける。まるで、複数のタスクが非同期で処理されているかのようだ。

「白い馬、お願いだ、生きて戻ってきてくれ！」スーホは、白い馬の後ろ姿を見つめながら祈った。

import asyncio

async def 白い馬の逃亡():
    print("白い馬は走り出した！")
    await asyncio.sleep(1)  # 矢を避ける時間
    print("矢が飛んでくる！")
    await asyncio.sleep(1)  # さらに逃げる時間
    print("スーホのもとへ急ぐ！")
    return "草原"

async def スーホの祈り():
    print("白い馬、生きて戻ってきてくれ！")
    await asyncio.sleep(2) # 白い馬の帰りを待つ
    return "祈り"

async def main():
    逃亡 = asyncio.create_task(白い馬の逃亡())
    祈り = asyncio.create_task(スーホの祈り())

    print("殿様の兵士たちが追いかけてくる！")

    逃亡_結果 = await 逃亡
    祈り_結果 = await 祈り

    print(f"白い馬は{逃亡_結果}へ、スーホは{祈り_結果}を捧げた。")


asyncio.run(main())

兵士隊長が叫ぶ。「逃がすな！殿様の命令だ！」

しかし、白い馬は速かった。兵士たちは、白い馬の速さに翻弄され、次第に追いつけなくなっていく。非同期処理によって、白い馬の逃亡と兵士たちの追跡は同時に行われている。まるで別々のスレッドで処理されているかのように、互いに干渉することなく、それぞれのタスクが実行されているのだ。

白い馬は、何度も矢を受けながらも、走り続けた。スーホの待つ草原を目指して。

悲しみと決意～Kafkaトピックの構築～

白い馬は、血を流し、よろめきながら、スーホのもとへたどり着いた。矢が白い馬の体に深く突き刺さっている。

「白い馬！しっかりしろ！」スーホは、白い馬を抱きしめ、必死に呼びかけた。

「スーホ…ありがとう…」白い馬は、かすれた声で whisper し、スーホの腕の中で静かに息を引き取った。

「白い馬…！」スーホの悲しみは、草原の風のように、どこまでも広がっていった。

数日後、深い悲しみに暮れるスーホの夢に、白い馬が現れた。「悲しむな、スーホ。私の体で楽器を作りなさい。そうすれば、私はいつもあなたのそばにいることができる」

「白い馬…君の体で…？」

「そうだ。私の皮で胴体を、骨でネックを、毛で弦を作りなさい。そして、私の魂をその楽器に吹き込みなさい」

スーホは、夢の中で白い馬の言葉を聞き、涙を流した。そして、白い馬の皮、骨、毛を使って馬頭琴を作ることを決意した。まるで、Apache Kafkaのトピックを構築するように、白い馬の体のパーツを構成要素として、新たな命を吹き込むのだ。

from kafka import KafkaProducer

async def 馬頭琴を作る(皮, 骨, 毛):
    producer = KafkaProducer(bootstrap_servers='localhost:9092') # Kafkaブローカーへの接続

    胴体 = await 皮を張る(皮)
    ネック = await 骨を削る(骨)
    弦 = await 毛を撚る(毛)

    馬頭琴 = {"胴体": 胴体, "ネック": ネック, "弦": 弦} # メッセージを辞書型で作成

    future = producer.send('白い馬の魂', value=馬頭琴) # トピック'白い馬の魂'にメッセージを送信
    result = future.get(timeout=60) # 送信結果を取得

    print(f"馬頭琴が完成しました: {result}")

    producer.close()


async def 皮を張る(皮):
    # ... 皮を張る処理 ...
    return "胴体完成"

async def 骨を削る(骨):
    # ... 骨を削る処理 ...
    return "ネック完成"

async def 毛を撚る(毛):
    # ... 毛を撚る処理 ...
    return "弦完成"

馬頭琴の制作は、Kafkaの設定や実装を意味する。トピック"白い馬の魂"は、白い馬の記憶と魂を永遠に保存するための場所なのだ。スーホは、白い馬の魂を胸に、馬頭琴を作り始めた。

草原に響く旋律～データパイプラインの完成～

スーホは、白い馬の皮を丁寧に鞣し、胴体を作り始めた。「白い馬、君の温もりを、この手に感じながら作るよ」

白い馬の骨は、硬く、そして美しかった。スーホは、その骨を一本一本、丁寧に削り、ネックを作り上げた。「白い馬、君の強さを、この楽器に込めたい」

最後に、白い馬の美しい毛を撚り、弦を作った。「白い馬、君の魂を、この弦に乗せて、草原に響かせたい」

from kafka import KafkaConsumer
import asyncio

async def 馬頭琴を奏でる(馬頭琴):
    consumer = KafkaConsumer('白い馬の魂', bootstrap_servers='localhost:9092')

    for message in consumer:
        魂 = message.value
        print(f"白い馬の魂を受け取りました：{魂}")
        await asyncio.sleep(1) # 音を響かせる
        print("草原に美しい旋律が響き渡る...")

async def main():
    馬頭琴 = {"胴体": "完成", "ネック": "完成", "弦": "完成"}
    await 馬頭琴を奏でる(馬頭琴)

asyncio.run(main())

こうして完成した馬頭琴は、まるで生きているかのようだった。スーホは、震える手で馬頭琴を構え、弓を弦に滑らせた。

「白い馬…聴いてくれ…これは、君への歌だ…」

馬頭琴から流れる音色は、美しく、そして力強かった。それは、白い馬の魂の叫びであり、スーホの白い馬への深い愛情の表現だった。音色は、Kafkaのデータパイプラインを流れるデータのように、草原を駆け巡り、全てを包み込んだ。

殿様は、その音色を聞き、驚き、そして恐怖に震えた。「何だ、あの音は…まるで、白い馬の魂が…！」

スーホは、馬頭琴を奏で続け、白い馬の魂を草原に響かせた。そして、静かに言った。「白い馬、これで君は永遠に、この草原で生き続けるんだ」

それから数百年後、スーホの子孫を名乗るコンサルタントが、Kafkaクラスタの安定稼働を誇らしげにプレゼンしていた。彼は、高可用性とスケーラビリティを熱く語り、集まった顧客たちは深く頷いていた。しかし、誰も白い馬の魂のことは、もう覚えていなかった。皮肉なことに、安定稼働の裏で、白い馬の物語は忘れ去られ、Kafkaは今日も粛々と大量のデータを処理し続けているのだった。