This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new baaddb8ee [CELEBORN-822][DOC] Introduce a quick start guide for
running Apache Flink with Apache Celeborn
baaddb8ee is described below
commit baaddb8ee812b0f65349fe3e6ffb0ad67a8ac1c4
Author: SteNicholas <[email protected]>
AuthorDate: Wed Aug 30 21:38:03 2023 +0800
[CELEBORN-822][DOC] Introduce a quick start guide for running Apache Flink
with Apache Celeborn
### What changes were proposed in this pull request?
Introduce a quick start guide for running Apache Flink with Apache Celeborn
to help Flink users to run with Celeborn.
### Why are the changes needed?
There is no quick start guide for running Apache Flink with Apache Celeborn.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
None.
Closes #1868 from SteNicholas/CELEBORN-822.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: zky.zhoukeyong <[email protected]>
---
.../plugin/flink/buffer/PartitionSortedBuffer.java | 3 +-
docs/README.md | 38 ++++++++++++++++++++--
docs/deploy.md | 2 +-
3 files changed, 38 insertions(+), 5 deletions(-)
diff --git
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
index 03beb5f9d..1b40508c9 100644
---
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
+++
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
@@ -391,8 +391,7 @@ public class PartitionSortedBuffer implements SortBuffer {
@Override
public void finish() {
- checkState(
- !isFinished, "com.alibaba.flink.shuffle.plugin.transfer.SortBuffer is
already finished.");
+ checkState(!isFinished, SortBuffer.class.getCanonicalName() + " is already
finished.");
isFinished = true;
diff --git a/docs/README.md b/docs/README.md
index c260ebca2..8bdf3b1c1 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -20,7 +20,7 @@ license: |
---
Quick Start
===
-This documentation gives a quick start guide for running Apache Spark with
Apache Celeborn(Incubating).
+This documentation gives a quick start guide for running Apache Spark/Flink
with Apache Celeborn(Incubating).
### Download Celeborn
Download the latest Celeborn binary from the [Downloading
Page](https://celeborn.apache.org/download/).
@@ -51,7 +51,7 @@ cd $CELEBORN_HOME
```
You should see `Master`'s ip:port in the log:
```log
-INFO [main] NettyRpcEnvFactory: Starting RPC Server [MasterSys] on
192.168.2.109:9097
+INFO [main] NettyRpcEnvFactory: Starting RPC Server [MasterSys] on
192.168.2.109:9097 with advisor endpoint 192.168.2.109:9097
```
#### Start Worker
Use the Master's IP and Port to start Worker:
@@ -63,6 +63,7 @@ You should see the following message in Worker's log:
```log
INFO [main] MasterClient: connect to master 192.168.2.109:9097.
INFO [main] Worker: Register worker successfully.
+INFO [main] Worker: Worker started.
```
And also the following message in Master's log:
```log
@@ -115,3 +116,36 @@ INFO [dispatcher-event-loop-9] Controller: Reserved 10
primary location and 0 re
INFO [dispatcher-event-loop-8] Controller: Start commitFiles for
local-1690000152711-0
INFO [async-reply] Controller: CommitFiles for local-1690000152711-0 success
with 10 committed primary partitions, 0 empty primary partitions, 0 failed
primary partitions, 0 committed replica partitions, 0 empty replica partitions,
0 failed replica partitions.
```
+
+## Start Flink with Celeborn
+#### Copy Celeborn Client to Flink's lib
+Celeborn release binary contains clients for Flink 1.14.x, Flink 1.15.x and
Flink 1.17.x, copy the corresponding client jar into Flink's
+`lib/` directory:
+```shell
+cp $CELEBORN_HOME/flink/<Celeborn Client Jar> $FLINK_HOME/lib/
+```
+#### Add Celeborn configuration to Flink's conf
+Set `shuffle-service-factory.class` to Celeborn's ShuffleServiceFactory in
Flink configuration file:
+```shell
+cd $FLINK_HOME
+vi conf/flink-conf.yaml
+```
+```properties
+shuffle-service-factory.class:
org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory
+```
+Then deploy the example word count job to the running cluster:
+```shell
+cd $FLINK_HOME
+
+./bin/flink run -Dexecution.runtime-mode=BATCH examples/streaming/WordCount.jar
+```
+During the Flink Job, you should see the following message in Celeborn
Master's log:
+```log
+Master: Offer slots successfully for 1 reducers of local-1690000152711-0 on 1
workers.
+```
+And the following message in Celeborn Worker's log:
+```log
+INFO [dispatcher-event-loop-4] Controller: Reserved 1 primary location and 0
replica location for local-1690000152711-0
+INFO [dispatcher-event-loop-3] Controller: Start commitFiles for
local-1690000152711-0
+INFO [async-reply] Controller: CommitFiles for local-1690000152711-0 success
with 1 committed primary partitions, 0 empty primary partitions, 0 failed
primary partitions, 0 committed replica partitions, 0 empty replica partitions,
0 failed replica partitions.
+```
\ No newline at end of file
diff --git a/docs/deploy.md b/docs/deploy.md
index cba476336..369c8cffe 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -190,7 +190,7 @@ spark.sql.adaptive.skewJoin.enabled true
Copy $CELEBORN_HOME/flink/*.jar to $FLINK_HOME/lib/
### Flink Configuration
-TO use Celeborn, following flink configurations should be added.
+To use Celeborn, following flink configurations should be added.
```properties
shuffle-service-factory.class:
org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory
celeborn.master.endpoints: clb-1:9097,clb-2:9097,clb-3:9097
