StrmVol存储卷：如何解锁K8s对象存储海量小文件访问性能新高度？

apiVersion: v1
kind: PersistentVolume
metadata:
  name: pv-strmvol
spec:
  capacity:
  # 挂载的OSS挂载点下最高可存储16 TiB数据。
    storage: 20Gi
  # 仅支持ReadOnlyMany访问模式。
  accessModes:
    - ReadOnlyMany
  persistentVolumeReclaimPolicy: Retain
  csi:
    driver: strmvol.csi.alibabacloud.com
    volumeHandle: pv-strmvol
    nodeStageSecretRef:
      name: strmvol-secret
      namespace: default
    volumeAttributes:
      bucket: imagenet
      path: /data
      url: oss-cn-hangzhou-internal.aliyuncs.com
      directMode: "false"
      resourceLimit: "4c8g"
---
kind: PersistentVolumeClaim
apiVersion: v1
metadata:
  name: pvc-strmvol
  namespace: default
spec:
  accessModes:
    - ReadOnlyMany
  resources:
    requests:
      storage: 20Gi
  volumeName: pv-strmvol

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: distributed-imagenet-training-
spec:
  entrypoint: main
  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
          - matchExpressions:
              - key: "node-type"
                operator: In
                values:
                  - "argo"
  volumes:
  - name: pvc-volume
    persistentVolumeClaim:
      claimName: pvc-strmvol
  templates:
  - name: main
    steps:
    - - name: list-shards
        template: list-imagenet-shards
    - - name: parallel-processing
        template: process-shard
        arguments:
          parameters:
            - name: paths
              value: "{{item}}"
        withParam: "{{steps.list-shards.outputs.result}}"
    - - name: calculate-statistics
        template: calculate-averages
  - name: list-imagenet-shards
    script:
      image: mirrors-ssl.aliyuncs.com/python:latest
      command: [python]
      source: |
        import subprocess
        import json
        output = subprocess.check_output("ls /mnt/data", shell=True, text=True)
        files = [f for f in output.split('\n') if f]
        print(json.dumps(files, indent=2))
      volumeMounts:
        - name: pvc-volume
          mountPath: /mnt/data
  - name: process-shard
    inputs:
      parameters:
      - name: paths
    container:
      image: alibaba-cloud-linux-3-registry.cn-hangzhou.cr.aliyuncs.com/alinux3/alinux3:latest
      command: [/bin/bash, -c]
      args:
      - |
        yum install -y parallel
        SHARD_JSON="/mnt/data/{{inputs.parameters.paths}}"
        SHARD_NUM="{{inputs.parameters.paths}}"
        START_TIME=$(date +%s)
        echo "Processing shard $SHARD_JSON"
        find "$SHARD_JSON" -maxdepth 1 -name "*.JPEG" -print0 | parallel -0 -j4 'cp {} /dev/null'
        END_TIME=$(date +%s)
        ELAPSED=$((END_TIME - START_TIME))
        mkdir -p /tmp/output
        echo $ELAPSED > /tmp/output/time_shard_${SHARD_NUM}.txt
      resources:
        requests:
          memory: "4Gi"
          cpu: "1000m"
        limits:
          memory: "4Gi"
          cpu: "2000m"
      volumeMounts:
        - name: pvc-volume
          mountPath: /mnt/data
    outputs:
      artifacts:
        - name: time_shard
          path: /tmp/output/time_shard_{{inputs.parameters.paths}}.txt
          oss:
            key: results/results-{{workflow.creationTimestamp}}/time_shard_{{inputs.parameters.paths}}.txt
          archive:
            none: {}
  - name: calculate-averages
    inputs:
      artifacts:
        - name: results
          path: /tmp/output
          oss:
            key: "results/results-{{workflow.creationTimestamp}}"
    container:
      image: registry-vpc.cn-beijing.aliyuncs.com/acs/busybox:1.33.1
      command: [sh, -c]
      args:
      - |
        echo "开始合并结果..."
        TOTAL_TIME=0
        SHARD_COUNT=0
        echo "各分片处理时间统计:"
        for time_file in /tmp/output/time_shard_*.txt; do
          TIME=$(cat $time_file)
          SHARD_ID=${time_file##*_}
          SHARD_ID=${SHARD_ID%.txt}
          echo "分片 ${SHARD_ID}: ${TIME} 秒"
          TOTAL_TIME=$((TOTAL_TIME + TIME))
          SHARD_COUNT=$((SHARD_COUNT + 1))
        done
        if [ $SHARD_COUNT -gt 0 ]; then
          AVERAGE=$((TOTAL_TIME / SHARD_COUNT))
          echo "--------------------------------"
          echo "总分片数量: $SHARD_COUNT"
          echo "总处理时间: $TOTAL_TIME 秒"
          echo "平均处理时间: $AVERAGE 秒/分片"
          echo "Average: $AVERAGE seconds" > /tmp/output/time_stats.txt
        else
          echo "错误：未找到分片时间数据"
          exit 1
        fi
    outputs:
      artifacts:
        - name: test-file
          path: /tmp/output/time_stats.txt
          oss:
            key: results/results-{{workflow.creationTimestamp}}/time_stats.txt
          archive:
            none: {}

开始合并结果...
各分片处理时间统计:
分片 1: 21 秒
分片 2: 21 秒
分片 3: 22 秒
分片 4: 21 秒
--------------------------------
总分片数量: 4
总处理时间: 85 秒
平均处理时间: 21 秒/分片