中国のEVメーカー小鵬汽車（シャオペン）が公開した人型ロボット「IRON」の新モデルが注目を集めている。同モデルは、同社が11月5日（現地時間）に中国広州市で開催したイベント「XPENG AI Day」で発表。滑らかに歩く姿を披露し、SNS上では「中に人いる？」などの反応を集めた。これを受け、小鵬汽車は6日、同モデルの“中身”を見せる動画を公開した。
中国の人型ロボ、自然すぎる歩き方に「中に人いる？」と話題に→開発企業が“中身”を公開、結果は…… - ITmedia AI＋
IRONは、小鵬汽車が2024年に公開した人型ロボットで、今回公開したのは第2世代のモデル。機械部分を皮膚を模した柔らかい素材で覆っており、全身に82の自由度を持つ。これにより、ファッションショーでモデルが披露する歩き方「キャットウォーク」などの高度な動作を実現したという。
中国の人型ロボ、自然すぎる歩き方に「中に人いる？」と話題に→開発企業が“中身”を公開、結果は…… - ITmedia AI＋
小鵬汽車の発表によると、IRONは動作性能が高いだけでなく、リチウムイオン電池に比べて安全性が高いとされる「全個体電池」を電源に採用。3000TOPSの演算性能を持つ自社製のAIチップを備えており、同社が開発したAIモデルで制御しているという。何CEOは「2026年末までに、高水準の人型ロボットの大量生産を目指す」としている。
中国の人型ロボ、自然すぎる歩き方に「中に人いる？」と話題に→開発企業が“中身”を公開、結果は…… - ITmedia AI＋
⇧ どこまで複雑な動きが可能なのかが分からないのですが、「AI」が「労働」を代替してくれる世界線が訪れるんですかね？
懸念としては、中国の企業ということで、
ノルウェーの公共交通機関運営会社であるRuterは、国内で運用している中国メーカー製の電動バスに「SIMカード」が搭載されていることを内部検査で発見しました。このSIMカードを用いると電動バスが遠隔から制御できる可能性があり、ノルウェー政府はバスのサイバーセキュリティリスクを再検討しています。
⇧ 上記のようなセキュリティリスクを抱えた製品になってくる可能性があるということですかね...
高い技術力があるというのに、誠に遺憾と言わざるを得ない...
2025年11月11日（火）追記：↓ ここから
何やら、
中国のバスメーカー・宇通のEVバス(電気バス)に、搭載されたSIMカード経由で外部からアクセス可能なセキュリティリスクがあることがノルウェーの事業者の調査で明らかになりました。この事態を受けて、イギリスなど他国でも同社のバスを運用する事業者が緊急調査を始めています。
ノルウェーの交通事業者・Ruterの調査により、宇通製のEVバスはSIMカード経由で外部から電源管理システムにアクセス可能であることが発覚しました。Ruterは、必要に応じて通信を切断したり、SIMカードを取り外したりすることで、バスをローカル環境で運用するよう変更しました。
オーストラリアでも、首都キャンベラを含む複数の州・準州・都市で宇通製EVバスが運行されています。オーストラリアで宇通のバス販売を行っているVDIは、100台超の導入があったものの、「オーストラリアでは、ソフトウェア更新はリモートではなく、サービスセンターで行うことが慣例になっている」と、セキュリティ面の問題がないことをアピールしたとのこと。
しかし、セキュリティ専門家でサイバーセキュリティセンター元所長のアラステア・マクギボン氏はABC Newsに対し、コネクテッドカーはソフトウェアやファームウェアの更新のためにメーカーと継続的にネット接続できる必要がある点を指摘し、今回の問題は「中国製バス」であるということではなく、「中国に本拠を置く企業は中国共産党の指示のもとにある」ということだと述べました。
⇧ 中国政府が関わっている問題であるとの見解。
まぁ、中国政府が介入してる事例があり過ぎるので、中国の企業は国外の市場開拓については茨の道というような気がしますな...
2025年11月11日（火）追記：↑ ここまで
Azure Blob Storage とは
公式のドキュメントによると、
Azure Blob Storage の概要
Azure Blob Storage は、Microsoft のクラウド用オブジェクト ストレージ ソリューションです。 Blob Storage は、テキスト データやバイナリ データなどの大量の非構造化データを格納するために最適化されています。 非構造化データとは、特定のデータ モデルや定義に従っていないデータであり、テキスト データやバイナリ データなどがあります。
https://learn.microsoft.com/ja-jp/azure/storage/blobs/storage-blobs-introduction
⇧ どんなデータでも格納できますと。
Blob Storage のリソース
Blob Storage には、3 種類のリソースがあります。
- ストレージ アカウント
- ストレージ アカウント内のコンテナー
- コンテナー内の BLOB
https://learn.microsoft.com/ja-jp/azure/storage/blobs/storage-blobs-introduction
次の図に、これらのリソースの関係を示します。
https://learn.microsoft.com/ja-jp/azure/storage/blobs/storage-blobs-introduction
⇧ とあるように、「コンテナー」配下に、ファイル群が格納されていく構造になっている。
ちなみに、「コンテナー」の容量の上限については、
Blob Storage のターゲットをスケーリングする
https://learn.microsoft.com/ja-jp/azure/storage/blobs/scalability-targets
⇧ 上記によると、
『ストレージ アカウントの最大容量と同じ』
とある。
「ストレージ アカウント」のドキュメントによると、
Standard ストレージ アカウントとディスク アクセス リソースのターゲットをスケーリングする
2 Azure Storage の Standard アカウントの場合、要求による容量制限とイングレスおよびエグレス制限の引き上げがサポートされます。 アカウント制限の引き上げを希望する場合は、Azure サポートにお問い合わせください。
https://learn.microsoft.com/ja-jp/azure/storage/common/scalability-targets-standard-account
⇧ とあり、
『5 PiB 2』
とある。
「ペピバイト（PiB：PepiByte）」はというと、
ペビバイト (pebibyte) とはコンピュータの容量や記憶装置の大きさをあらわす情報の単位の一つ。PiBと略記される。
- 1 PiB = 250 B = 1,125,899,906,842,624 B
SI接頭辞ペタがついたペタバイトと同じく250バイトを表す言葉である。しかし本来SI接頭語であり1015を表すペタを使ったペタバイトは1,000,000,000,000,000バイトに用いられることもある。このような混乱を避けるためIECが決めた二進接頭辞を取り入れペビバイト = 250バイトとしており、250バイトという意味ではこの呼び名が推奨される。ペタバイトと違い千兆バイトという意味で使えない。
⇧ とあり、まぁ、相当に大容量ではある。
Azure Blob Storageの大量のファイル群をダウンロードしたいのだが...
で、残念ながら、ファイルの内容を確認するには、「ダウンロード」が必要なのだが、保存期間を設けているにしろ、膨大な数のファイルが保存されていること、あるあるですと。
で、「ChatGPT」氏に質問してみたところ、以下のような回答が返ってきた。
✅ この構成の特徴
-
VMディスク容量500GBでも安全に処理
-
小ファイル多数でもOutOfMemory回避
-
AzCopy並列度はメモリ量に応じて自動調整
-
上半期/下半期/月単位/年単位の4種類のAPIエンドポイント対応
-
FastAPI経由で外部から非同期トリガー可能
-
VM→ローカルへの安全転送
■プロジェクト構成
azure_download_project/
├─ main.py # FastAPI起動
├─ controller.py # Controller(FastAPI) + 並列制御
├─ splitter.py # 年/月単位ファイル列挙・バッチ分割
├─ azcopy_job.py # AzCopyサブプロセス実行
├─ archiver.py # VM上でバッチ単位アーカイブ作成
├─ transfer_mgr.py # VM→ローカル転送
├─ transfer_config.py # VM容量・メモリに応じたバッチサイズ/並列度調整
1️⃣ transfer_config.py
import shutil
import psutil
class TransferConfig:
@staticmethod
def get_vm_free_space(path="/mnt/vm_storage"):
"""VMの空き容量を取得"""
usage = shutil.disk_usage(path)
return usage.free # バイト
@staticmethod
def get_max_workers(default=8):
"""メモリに応じてAzCopy並列数を調整"""
mem = psutil.virtual_memory()
if mem.total < 4*1024**3:
return 2
elif mem.total < 8*1024**3:
return 4
else:
return default
@staticmethod
def get_batch_size(vm_path="/mnt/vm_storage", safety_ratio=0.2):
"""VM空き容量の安全比率に応じてバッチサイズを決定"""
free_space = TransferConfig.get_vm_free_space(vm_path)
batch_size = int(free_space * safety_ratio)
return max(batch_size, 1*1024**3) # 最低1GB
2️⃣ splitter.py
import os
class JobSplitter:
def __init__(self, base_dir="/mnt/vm_storage"):
self.base_dir = base_dir
def list_files(self, year, month=None):
"""指定年月のファイルをジェネレータで返す"""
path = os.path.join(self.base_dir, year)
if month:
path = os.path.join(path, f"{month:02d}")
for root, dirs, files in os.walk(path):
for f in files:
yield os.path.join(root, f)
def split_jobs(self, file_iter, batch_size_bytes=10*1024**3):
"""ファイルリストをバッチサイズごとに分割"""
batch = []
batch_size = 0
for f in file_iter:
size = os.path.getsize(f)
if batch_size + size > batch_size_bytes and batch:
yield batch
batch = []
batch_size = 0
batch.append(f)
batch_size += size
if batch:
yield batch
3️⃣ azcopy_job.py
import subprocess
class AzCopyJob:
def __init__(self, files):
self.files = files
def run(self):
"""AzCopyでVM→ローカルに安全にコピー"""
for f in self.files:
dest = f.replace("/mnt/vm_storage", "/local_storage")
subprocess.run(["azcopy", "copy", f, dest, "--recursive"], check=True)
4️⃣ archiver.py
import shutil
import os
class Archiver:
def create_archive(self, file_list, batch_index, year, label):
"""バッチごとにZIPアーカイブを作成"""
archive_name = f"/mnt/vm_storage/archives/{year}_{label}_batch{batch_index}.zip"
os.makedirs(os.path.dirname(archive_name), exist_ok=True)
# 1バッチだけ圧縮
batch_dir = os.path.commonpath(file_list)
shutil.make_archive(archive_name.replace(".zip", ""), 'zip', root_dir=batch_dir)
return archive_name
5️⃣ transfer_mgr.py
import shutil
import os
class TransferManager:
def transfer_archive(self, archive_path):
"""VM上のアーカイブをローカルPCに移動"""
local_dest = archive_path.replace("/mnt/vm_storage", "/local_storage")
os.makedirs(os.path.dirname(local_dest), exist_ok=True)
shutil.move(archive_path, local_dest)
6️⃣ controller.py
from fastapi import APIRouter, BackgroundTasks
from splitter import JobSplitter
from archiver import Archiver
from transfer_mgr import TransferManager
from azcopy_job import AzCopyJob
from transfer_config import TransferConfig
from concurrent.futures import ThreadPoolExecutor, as_completed
router = APIRouter()
class DownloadController:
def __init__(self, vm_base_dir="/mnt/vm_storage"):
self.running = False
self.splitter = JobSplitter(vm_base_dir)
self.archiver = Archiver()
self.transfer_mgr = TransferManager()
self.vm_base_dir = vm_base_dir
def _process_files(self, file_iter, year, label):
batch_size_bytes = TransferConfig.get_batch_size(self.vm_base_dir)
max_workers = TransferConfig.get_max_workers()
for batch_index, job_files in enumerate(self.splitter.split_jobs(file_iter, batch_size_bytes=batch_size_bytes)):
# AzCopy並列実行
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [executor.submit(AzCopyJob([f]).run) for f in job_files]
for _ in as_completed(futures):
pass
# バッチをアーカイブ
archive_path = self.archiver.create_archive(job_files, batch_index, year, label)
# ローカルへ転送
self.transfer_mgr.transfer_archive(archive_path)
def _start_download(self, file_iter, year, label, background_tasks: BackgroundTasks):
if self.running:
return {"status": "already running"}
self.running = True
def task():
try:
self._process_files(file_iter, year, label)
finally:
self.running = False
background_tasks.add_task(task)
return {"status": f"Download started for {label} of year {year}"}
# 年度
@router.post("/download/year/{year}/")
def download_year(self, year: str, background_tasks: BackgroundTasks):
file_iter = self.splitter.list_files(year)
return self._start_download(file_iter, year, "year", background_tasks)
# 月
@router.post("/download/year/{year}/month/{month}/")
def download_month(self, year: str, month: str, background_tasks: BackgroundTasks):
file_iter = self.splitter.list_files(year, month)
return self._start_download(file_iter, year, f"month_{month}", background_tasks)
# 上半期（1-6月）
@router.post("/download/year/{year}/first-half/")
def download_first_half(self, year: str, background_tasks: BackgroundTasks):
def file_gen():
for month in range(1,7):
yield from self.splitter.list_files(year, f"{month:02d}")
return self._start_download(file_gen(), year, "first_half", background_tasks)
# 下半期（7-12月）
@router.post("/download/year/{year}/second-half/")
def download_second_half(self, year: str, background_tasks: BackgroundTasks):
def file_gen():
for month in range(7,13):
yield from self.splitter.list_files(year, f"{month:02d}")
return self._start_download(file_gen(), year, "second_half", background_tasks)
7️⃣ main.py
from fastapi import FastAPI
from controller import DownloadController, router
app = FastAPI()
controller = DownloadController()
app.include_router(router)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
⇧ 動作確認はできていないのだが、
この構成で、1TB・数千万〜数十億の小ファイルでも、VMを経由して高速かつ安全にローカルPCにダウンロード可能です。
必要であれば、この構成での処理時間の概算シミュレーションも出せます。
出しますか？
と、「ChatGPT」氏の考えでは、動作することが確定しているっぽいので、「処理時間の概算」をお願いしてみた。
処理時間の概算
■前提条件
-
データ量：1TB（約1,000,000 MB）
-
ファイルサイズ：平均10KB（小さいファイル多数）
-
ファイル数：約100,000,000 個
-
VMディスク容量：500GB
-
VM→ローカル転送速度：1Gbps（実効125 MB/s）
-
AzCopy並列度：4〜8（CPU/メモリ依存）
-
圧縮率：小ファイルなのでZIP圧縮で50%減と仮定（非常に概算）
■ステップごとの時間概算
-
-
バッチサイズ：100GB
-
1バッチに約10,000,000個のファイル
-
小ファイル多数なので、I/Oオーバーヘッドあり
-
実測AzCopy平均速度（小ファイル多数）は50MB/s程度
-
バッチ100GB転送時間：約100GB ÷ 50MB/s ≈ 2,048秒 ≈ 34分
-
-
アーカイブ作成（ZIP）
-
小ファイルをまとめて圧縮
-
CPU依存。1バッチ100GBで50%圧縮の場合：
-
おおよそ30〜60分（CPU2〜4コア時の目安）
-
-
VM→ローカル転送
-
圧縮後50GB
-
転送速度1Gbps → 125 MB/s
-
50GB ÷ 125 MB/s ≈ 400秒 ≈ 7分
-
■1TB全体を処理する場合
-
1TB ÷ 100GB/バッチ = 10バッチ
-
バッチ単位の総時間：
-
AzCopy：34分 × 10 = 340分 ≈ 5.7時間
-
ZIP圧縮：45分 × 10 = 450分 ≈ 7.5時間
-
転送：7分 × 10 = 70分 ≈ 1.2時間
-
合計概算 ≈ 14.4時間
※小ファイル多数のI/Oオーバーヘッド、CPU、VMスペックによって増減します。
-
並列度を上げるとAzCopy部分は数倍高速化可能
-
バッチサイズを減らすとVMディスク不足リスクは減るが、オーバーヘッド増加で総時間はやや増加
💡 まとめ
-
VMを介した分割ダウンロードは安全で、OutOfMemoryは発生しません
-
並列AzCopy + バッチ処理 + 圧縮 + 転送の組み合わせで、大規模小ファイルの処理も現実的
-
実際の速度は、ファイル数とVMスペック、ネットワーク帯域に大きく依存します
う～む...
「1TiB」の容量のファイル群をダウンロードする処理に、半日以上かかるのは一般的なのだろうか？
実際問題、これだけ処理に時間がかかるとするならば、
- ダウンロード用の圧縮ファイルを作成する処理
- 圧縮ファイルをダウンロードする処理
を分けるしか無いですな...
別々の日に実施するような仕組みにしないと厳しいということですな...
兎にも角にも、「AI」の回答だから、「ユーザーエクスペリエンス（UX：User eXperience）」まで考慮できないのは致し方ないとは言え、容量が「1TiB」ともなるとエグい時間かかりますな...
まぁ、「AI」の見積もりが全く見当外れということもあるので何とも言えませんが...
ちなみに、公式のドキュメントによると、
オフライン転送またはネットワーク転送
大規模なデータセットは、データの TB が少ないから数 TB のデータがあることを意味します。 ネットワーク帯域幅がない、ネットワークが遅い、または信頼性が低い。 また、
- インターネット サービス プロバイダー (ISP) からのネットワーク転送のコストによって制限されます。
- セキュリティまたは組織のポリシーでは、機密データを処理するときに送信接続を許可しません。
https://learn.microsoft.com/ja-jp/azure/storage/common/storage-solution-large-dataset-low-network
物理デバイスが適切なオプションかどうかを確認するには、次の表を使用します。 これには、さまざまな使用可能な帯域幅 (90% 使用率を想定) のネットワーク データ転送の予測時間が表示されます。 ネットワーク転送が遅すぎると予測される場合は、物理デバイスを使用する必要があります。
https://learn.microsoft.com/ja-jp/azure/storage/common/storage-solution-large-dataset-low-network
⇧ とあるのだが、ファイル数が膨大な場合についての処理時間の記載が無いのことから、何とも言えないのよな...
どちらにしろ、
- ダウンロード用の圧縮ファイルを作成する処理
- 圧縮ファイルをダウンロードする処理
は、分ける設計にする必要があるとは思いますが...
「1. ダウンロード用の圧縮ファイルを作成する処理」については、無駄に何回も実行されないようにする制御が必要になってくるとは思いますが、設計が難しいですな...
そもそも、
- Azure Blob Storage
- Azure 仮想マシン
がディレクトリをマウントしているのならば、「AzCopy」を使わなくても良さそうなのよな...
ということで、「ChatGPT」氏に、ファイルを圧縮する方法について質問してみたところ、以下のような回答が返ってきた。
🔹まとめ（速度・圧縮率バランス）
| ツール | 並列化 | 速度 | 圧縮率 | 互換性 |
|---|---|---|---|---|
pigz |
✅ | ★★★★★ | ★★☆☆☆ | gzip互換 |
pbzip2 |
✅ | ★★★☆☆ | ★★★★☆ | bzip2互換 |
zstd |
✅ | ★★★★★ | ★★★★☆ | 新しめ（tar対応） |
xz |
❌（標準は単一スレッド） | ★☆☆☆☆ | ★★★★★ | xz互換 |
🔹 各コマンドのデフォルト拡張子
| ツール | コマンド例 | 生成されるファイル名 | 備考 |
|---|---|---|---|
| gzip / pigz | pigz file.txt |
file.txt.gz |
.gz 拡張子が自動で付く |
| bzip2 / pbzip2 | pbzip2 file.txt |
file.txt.bz2 |
.bz2 拡張子が付く |
| xz | xz file.txt |
file.txt.xz |
.xz |
| zstd | zstd file.txt |
file.txt.zst |
.zst |
| zip | zip archive.zip file1 file2 |
archive.zip |
明示的に指定する |
| tar + 圧縮 | tar -czf archive.tar.gz dir/ |
archive.tar.gz |
-c(作成) + -z(gzip) |
| tar + zstd | tar -I zstd -cf archive.tar.zst dir/ |
archive.tar.zst |
tarの拡張子は手動で指定 |
🔹 まとめ（覚え方）
| 圧縮形式 | 拡張子 | コマンド |
|---|---|---|
| gzip系 | .gz |
gzip, pigz |
| bzip2系 | .bz2 |
bzip2, pbzip2 |
| xz系 | .xz |
xz |
| zstd | .zst |
zstd |
| zip | .zip |
zip |
| tar + 圧縮 | .tar.gz, .tar.zst, .tar.xz |
tar -czf, tar -I zstd, etc |
✅ おすすめ構成
-
とにかく速く →
zstd -T0 -3 -
速度と圧縮率の両立 →
zstd -T0 -10 -
従来のgzip互換が必要 →
pigz -9 -
一括大量ファイル →
find ... | parallel pigz
う～む...
とりあえず、「AI」の回答が正しいと仮定するならば、「zstd」を使うのが無難という感じになるんかね。
ただ、圧縮後の「拡張子」的に、「zstd」は「Windows」環境とか対応していないような気がするんよな...
何やら、
⇧ 上記サイト様によりますと、「Windows」環境であっても「zstd」をインストールしてもらえるならば、イケるらしいですと。
「WSL 2（Windows Subsystem for Linux 2）」とかで「Linux」環境を用意してもらうという手もあるとは思いますが...
まぁ、「Windows」環境は、どうしても、ひと手間は必要になってきてしまうのよね...
By the way、圧縮ツールについては、
⇧ 上記サイト様が整理してくださっておりました。
毎度モヤモヤ感が半端ない…
今回はこのへんで。
