Checkpoint-uri Apache Flink pe S3 și pe stocare compatibilă cu S3

  • 1 minute de citit
  • 2 exemple de cod

Disponibil și în English

Pe această pagină

Salut,

Recent mi-a scris cineva de la Yahoo despre un vechi thread pe care îl deschisesem pe lista de discuții a utilizatorilor Apache Flink. I-am răspuns pe e-mail, dar am decis să transform răspunsul într-un articol de blog, pentru că poate ajuta și pe alții.

Email

Salut,

Am reușit să-l fac să funcționeze după ce am mai experimentat. Problema a fost în principal o neînțelegere: nu știam oficial ce metodă de autentificare foloseam în AWS. Foloseam doar s3://

Iată opțiunile noastre de configurare:

Pe stocarea compatibilă cu S3:

fs.s3a.access.key: ""fs.s3a.secret.key: ""fs.s3a.connection.ssl.enabled: "false"fs.s3a.endpoint: "ceph-mcr-1.xxx.xxx.xxx:xxx"fs.s3a.list.version: "1"s3.path.style.access: "true"containerized.master.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.13.2.jar"containerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.13.2.jar"state.backend: "rocksdb"state.backend.incremental: "true"state.checkpoints.dir: "s3://bucket-name/checkpoints/$cluster_name$/"state.savepoints.dir: "s3://bucket-name/savepoints/$cluster_name$/"

Pe S3 cu AWS:

fs.s3a.aws.credentials.provider: "com.amazonaws.auth.WebIdentityTokenCredentialsProvider",containerized.master.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.13.2.jar",containerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.13.2.jar",state.backend: "rocksdb",state.backend.incremental: "true",state.checkpoints.dir: "s3://xxx/checkpoints/$cluster_name$/",state.savepoints.dir: "s3://xxx/savepoints/$cluster_name$/"

fs.s3a.aws.credentials.provider era metoda de autentificare (furnizorul de credențiale) care ne lipsea. Nu apare în documentația plugin-ului Hadoop[2], dar apare în documentația AWS Java SDK[3][4]. AWS montează secretele în pod-urile Flink, așa că folosirea acestui furnizor ar trebui să funcționeze fără configurare suplimentară.

Ține cont că flink-s3-fs-hadoop-1.13.2.jar trebuie adaptat la versiunea ta de Flink. $cluster_name trebuie, de asemenea, înlocuit cu numele clusterului/deployment-ului tău.

Cam atât. Atașez și documentația Flink S3[1] la e-mail. Mulțumesc că ai scris! Sper că îți dai seama până la urmă!

Cu stimă,

Denis

docs/deployment/filesystems/s3/[1]
hadoop-aws/tools/hadoop-aws/index.html#S3A[2]
amazonaws/auth/AWSCredentialsProvider.html[3]
amazonaws/auth/WebIdentityTokenCredentialsProvider.html[4]

Ca observație, dacă folosești Flink Operator pentru a lansa job-ul Flink, poți seta variabilele de mediu în fișierul pod template în loc de flink-config.yaml.

Mulțumesc pentru lectură!