如何配置Flink以Ceph为存储后端？求相关配置用于性能测试

Hey there! I've helped a bunch of developers set up Flink with Ceph for persistent storage, so let's break this down step by step—including all the configs you need for performance testing.

前提准备 #

First, make sure your Ceph cluster is fully operational, and you've prepped the right storage resources:

• If you're using CephFS (great for file-based storage like checkpoints and savepoints): Create a CephFS filesystem, and grab your monitor addresses, auth ID, and keyring file path.
• If you're using RBD (RADOS Block Device) (good for block-level storage): Create an RBD image, then map and mount it to every node in your Flink cluster (so Flink can treat it like a local directory).

方案1：用 CephFS 作为 Flink 的文件系统后端 #

This is the most flexible approach since Flink works seamlessly with Hadoop-compatible file systems, and CephFS has a Hadoop client.

Step 1: Add Ceph dependencies to Flink #

Flink doesn't ship with Ceph clients out of the box, so drop the right JAR into your Flink lib directory:

• For CephFS: Grab the cephfs-hadoop JAR that matches your Hadoop version (Flink uses Hadoop's filesystem API under the hood).

Step 2: Update Flink's config file (conf/flink-conf.yaml) #

Add these lines to connect Flink to your CephFS cluster:

# Set CephFS as the default filesystem (optional—you can also use `ceph://` prefix in paths)
fs.default-scheme: ceph://<your-ceph-mon-host>:6789/

# CephFS client configs
fs.ceph.impl: org.apache.hadoop.fs.ceph.CephFileSystem
fs.ceph.mon.address: <mon-ip-1>:6789,<mon-ip-2>:6789  # List all your Ceph monitors
fs.ceph.auth.id: admin  # Your Ceph auth user ID
fs.ceph.auth.keyring: /etc/ceph/ceph.client.admin.keyring  # Path to your keyring (ensure Flink can read it)
fs.ceph.data.pool: cephfs_data  # Name of your CephFS data pool
fs.ceph.metadata.pool: cephfs_metadata  # Name of your CephFS metadata pool

Step 3: Configure Flink state storage #

To store checkpoints, savepoints, and state in CephFS, update these state backend settings:

# Use the Filesystem state backend (best for CephFS)
state.backend: filesystem
# Path for checkpoints in CephFS
state.checkpoints.dir: ceph:///flink-checkpoints
# Path for savepoints
state.savepoints.dir: ceph:///flink-savepoints
# Optional: Enable incremental checkpoints to reduce data transfer
state.backend.incremental: true

方案2：用 RBD 块设备的简化配置 #

If you prefer using RBD, just mount the RBD image to a directory on every Flink node (e.g., /mnt/ceph-rbd), then point Flink's state paths to that local directory:

state.backend: filesystem
state.checkpoints.dir: file:///mnt/ceph-rbd/flink-checkpoints
state.savepoints.dir: file:///mnt/ceph-rbd/flink-savepoints

性能测试优化配置 #

To get accurate performance metrics for Flink + Ceph, tweak these settings:

• Checkpoint tuning:

  # Adjust checkpoint interval based on your test scenario
  execution.checkpointing.interval: 1min
  # Allow concurrent checkpoints to boost write throughput
  execution.checkpointing.max-concurrent-checkpoints: 2
  # Extend timeout to avoid failures if Ceph is under load
  execution.checkpointing.timeout: 10min
  

• Ceph client optimizations:
  • Edit /etc/ceph/ceph.conf on Flink nodes:
    • Increase rbd_cache_size (e.g., 1073741824 for 1GB) to reduce disk I/O
    • For CephFS, set fs.ceph.read.ahead.bytes: 131072 to optimize read performance
• Flink resource tweaks:
  • Assign more heap memory to TaskManagers to handle state caching
  • Adjust job parallelism to match your Ceph cluster's throughput capacity

验证配置是否生效 #

1. Start your Flink cluster
2. Submit a test job (like the official WordCount example)
3. Trigger a checkpoint manually via the Flink Web UI
4. Check your Ceph cluster—you should see checkpoint/savepoint files in the paths you configured
5. Monitor the Flink Web UI's checkpoint metrics (e.g., completion time, success rate) to gauge performance

内容的提问来源于stack exchange，提问作者Vinod Gavhane