You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2022/12/29 03:52:16 UTC
[incubator-celeborn] branch branch-0.2 updated: [CELEBORN-172][Refactor] Load/Make snapshot use Protobuf serde (#1118)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.2
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.2 by this push:
new 2f4c162e [CELEBORN-172][Refactor] Load/Make snapshot use Protobuf serde (#1118)
2f4c162e is described below
commit 2f4c162e22066b1e7622c5e002976bc69149c6fa
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Thu Dec 29 11:51:14 2022 +0800
[CELEBORN-172][Refactor] Load/Make snapshot use Protobuf serde (#1118)
---
common/src/main/proto/TransportMessages.proto | 26 ++++
.../celeborn/common/meta/AppDiskUsageMetric.scala | 31 ++++-
.../apache/celeborn/common/util/PbSerDeUtils.scala | 76 ++++++++++-
.../master/clustermeta/AbstractMetaManager.java | 148 +++++++--------------
.../clustermeta/ha/MasterStateMachineSuiteJ.java | 31 ++++-
5 files changed, 206 insertions(+), 106 deletions(-)
diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index 2cf93798..07cd0db8 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -428,3 +428,29 @@ message PbResourceConsumption {
int64 hdfsBytesWritten = 3;
int64 hdfsFileCount = 4;
}
+
+message PbAppDiskUsage {
+ string appId = 1;
+ int64 estimatedUsage = 2;
+}
+
+message PbAppDiskUsageSnapshot {
+ int32 topItemCount = 1;
+ repeated PbAppDiskUsage topNItems = 2;
+ int64 startSnapShotTime = 3;
+ int64 endSnapshotTime = 4;
+}
+
+message PbSnapshotMetaInfo {
+ int64 estimatedPartitionSize = 1;
+ repeated string registeredShuffle = 2;
+ repeated string hostnameSet = 3;
+ repeated PbWorkerInfo blacklist = 4;
+ repeated PbWorkerInfo workerLostEvents = 5;
+ map<string, int64> appHeartbeatTime = 6;
+ repeated PbWorkerInfo workers = 7;
+ int64 partitionTotalWritten = 8;
+ int64 partitionTotalFileCount = 9;
+ repeated PbAppDiskUsageSnapshot appDiskUsageMetricSnapshots = 10;
+ optional PbAppDiskUsageSnapshot currentAppDiskUsageMetricsSnapshot = 11;
+}
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala b/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
index 6909489a..563243b0 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/AppDiskUsageMetric.scala
@@ -33,8 +33,8 @@ case class AppDiskUsage(var appId: String, var estimatedUsage: Long) {
}
class AppDiskUsageSnapShot(val topItemCount: Int) extends Logging with Serializable {
- val topNItems = new Array[AppDiskUsage](topItemCount)
- val startSnapShotTime = System.currentTimeMillis()
+ var topNItems = new Array[AppDiskUsage](topItemCount)
+ var startSnapShotTime = System.currentTimeMillis()
var endSnapShotTime: Long = _
def commit(): Unit = {
@@ -78,6 +78,25 @@ class AppDiskUsageSnapShot(val topItemCount: Int) extends Logging with Serializa
-1
}
+ def restoreFromSnapshot(array: Array[AppDiskUsage]): Unit = {
+ // Restored snapshots only contains values not null
+ for (i <- 0 until (topItemCount)) {
+ if (i < array.length) {
+ topNItems(i) = array(i)
+ } else {
+ topNItems(i) = null
+ }
+ }
+ }
+
+ override def equals(obj: Any): Boolean = {
+ obj.isInstanceOf[AppDiskUsageSnapShot] &&
+ obj.asInstanceOf[AppDiskUsageSnapShot].topItemCount == topItemCount &&
+ obj.asInstanceOf[AppDiskUsageSnapShot].startSnapShotTime == startSnapShotTime &&
+ obj.asInstanceOf[AppDiskUsageSnapShot].endSnapShotTime == endSnapShotTime &&
+ obj.asInstanceOf[AppDiskUsageSnapShot].topNItems.zip(topNItems).forall { case (x, y) => x == y }
+ }
+
override def toString(): String = {
val zoneId = ZoneId.systemDefault()
s"Snapshot " +
@@ -146,9 +165,13 @@ class AppDiskUsageMetric(conf: CelebornConf) extends Logging {
}
def restoreFromSnapshot(array: Array[AppDiskUsageSnapShot]): Unit = {
+ // Restored snapshots only contains values not null
for (i <- 0 until (snapshotCount)) {
- snapShots(i) = array(i)
+ if (i < array.length) {
+ snapShots(i) = array(i)
+ } else {
+ snapShots(i) = null
+ }
}
}
-
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index d81dbe65..e35725ab 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -19,15 +19,14 @@ package org.apache.celeborn.common.util
import java.util
import java.util.concurrent.ConcurrentHashMap
+import java.util.function.IntFunction
import scala.collection.JavaConverters._
import com.google.protobuf.InvalidProtocolBufferException
import org.apache.celeborn.common.identity.UserIdentifier
-import org.apache.celeborn.common.meta.DiskInfo
-import org.apache.celeborn.common.meta.FileInfo
-import org.apache.celeborn.common.meta.WorkerInfo
+import org.apache.celeborn.common.meta.{AppDiskUsage, AppDiskUsageSnapShot, DiskInfo, FileInfo, WorkerInfo}
import org.apache.celeborn.common.protocol._
import org.apache.celeborn.common.protocol.PartitionLocation.Mode
import org.apache.celeborn.common.protocol.message.ControlMessages.WorkerResource
@@ -318,4 +317,75 @@ object PbSerDeUtils {
}).asJava
}
+ def fromPbAppDiskUsage(pbAppDiskUsage: PbAppDiskUsage): AppDiskUsage = {
+ AppDiskUsage(pbAppDiskUsage.getAppId, pbAppDiskUsage.getEstimatedUsage)
+ }
+
+ def toPbAppDiskUsage(appDiskUsage: AppDiskUsage): PbAppDiskUsage = {
+ PbAppDiskUsage.newBuilder()
+ .setAppId(appDiskUsage.appId)
+ .setEstimatedUsage(appDiskUsage.estimatedUsage)
+ .build()
+ }
+
+ def fromPbAppDiskUsageSnapshot(
+ pbAppDiskUsageSnapShot: PbAppDiskUsageSnapshot): AppDiskUsageSnapShot = {
+ val snapShot = new AppDiskUsageSnapShot(pbAppDiskUsageSnapShot.getTopItemCount)
+ snapShot.startSnapShotTime = pbAppDiskUsageSnapShot.getStartSnapShotTime
+ snapShot.endSnapShotTime = pbAppDiskUsageSnapShot.getEndSnapshotTime
+ snapShot.restoreFromSnapshot(
+ pbAppDiskUsageSnapShot
+ .getTopNItemsList
+ .asScala
+ .map(fromPbAppDiskUsage)
+ .asJava
+ .stream()
+ .toArray(new IntFunction[Array[AppDiskUsage]]() {
+ override def apply(value: Int): Array[AppDiskUsage] = new Array[AppDiskUsage](value)
+ }))
+ snapShot
+ }
+
+ def toPbAppDiskUsageSnapshot(snapshots: AppDiskUsageSnapShot): PbAppDiskUsageSnapshot = {
+ PbAppDiskUsageSnapshot.newBuilder()
+ .setTopItemCount(snapshots.topItemCount)
+ .setStartSnapShotTime(snapshots.startSnapShotTime)
+ .setEndSnapshotTime(snapshots.endSnapShotTime)
+ // topNItems some value could be null
+ .addAllTopNItems(snapshots.topNItems.filter(_ != null).map(toPbAppDiskUsage).toList.asJava)
+ .build()
+ }
+
+ def toPbSnapshotMetaInfo(
+ estimatedPartitionSize: java.lang.Long,
+ registeredShuffle: java.util.Set[String],
+ hostnameSet: java.util.Set[String],
+ blacklist: java.util.Set[WorkerInfo],
+ workerLostEvent: java.util.Set[WorkerInfo],
+ appHeartbeatTime: java.util.Map[String, java.lang.Long],
+ workers: java.util.ArrayList[WorkerInfo],
+ partitionTotalWritten: java.lang.Long,
+ partitionTotalFileCount: java.lang.Long,
+ appDiskUsageMetricSnapshots: Array[AppDiskUsageSnapShot],
+ currentAppDiskUsageMetricsSnapshot: AppDiskUsageSnapShot): PbSnapshotMetaInfo = {
+ val builder = PbSnapshotMetaInfo.newBuilder()
+ .setEstimatedPartitionSize(estimatedPartitionSize)
+ .addAllRegisteredShuffle(registeredShuffle)
+ .addAllHostnameSet(hostnameSet)
+ .addAllBlacklist(blacklist.asScala.map(toPbWorkerInfo(_, true)).asJava)
+ .addAllWorkerLostEvents(workerLostEvent.asScala.map(toPbWorkerInfo(_, true)).asJava)
+ .putAllAppHeartbeatTime(appHeartbeatTime)
+ .addAllWorkers(workers.asScala.map(toPbWorkerInfo(_, true)).asJava)
+ .setPartitionTotalWritten(partitionTotalWritten)
+ .setPartitionTotalFileCount(partitionTotalFileCount)
+ // appDiskUsageMetricSnapshots can have null values,
+ // protobuf repeated value can't support null value in list.
+ .addAllAppDiskUsageMetricSnapshots(appDiskUsageMetricSnapshots.filter(_ != null)
+ .map(toPbAppDiskUsageSnapshot).toList.asJava)
+ if (currentAppDiskUsageMetricsSnapshot != null) {
+ builder.setCurrentAppDiskUsageMetricsSnapshot(
+ toPbAppDiskUsageSnapshot(currentAppDiskUsageMetricsSnapshot))
+ }
+ builder.build()
+ }
}
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
index c8ac89cb..75f18c26 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
@@ -20,6 +20,7 @@ package org.apache.celeborn.service.deploy.master.clustermeta;
import static org.apache.celeborn.common.protocol.RpcNameConstants.WORKER_EP;
import java.io.*;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -29,19 +30,19 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
+import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.identity.UserIdentifier;
-import org.apache.celeborn.common.meta.AppDiskUsageMetric;
-import org.apache.celeborn.common.meta.AppDiskUsageSnapShot;
-import org.apache.celeborn.common.meta.DiskInfo;
-import org.apache.celeborn.common.meta.WorkerInfo;
+import org.apache.celeborn.common.meta.*;
+import org.apache.celeborn.common.protocol.PbSnapshotMetaInfo;
import org.apache.celeborn.common.quota.ResourceConsumption;
import org.apache.celeborn.common.rpc.RpcAddress;
import org.apache.celeborn.common.rpc.RpcEnv;
+import org.apache.celeborn.common.util.PbSerDeUtils;
import org.apache.celeborn.common.util.Utils;
public abstract class AbstractMetaManager implements IMetadataHandler {
@@ -243,63 +244,21 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
* @throws IOException
*/
public void writeMetaInfoToFile(File file) throws IOException, RuntimeException {
- ObjectOutputStream out =
- new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(file)));
-
- out.writeLong(estimatedPartitionSize);
- // write registeredShuffle
- writeSetMetaToFile(registeredShuffle, out);
-
- // write hostnameSet
- writeSetMetaToFile(hostnameSet, out);
-
- // write blacklist
- writeSetMetaToFile(blacklist, out);
-
- // write workerLost events
- writeSetMetaToFile(workerLostEvents, out);
-
- // write application heartbeat time
- out.writeInt(appHeartbeatTime.size());
- appHeartbeatTime.forEach(
- (app, time) -> {
- try {
- out.writeObject(app);
- out.writeLong(time);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
-
- // write workerInfo
- out.writeInt(workers.size());
- workers.forEach(
- workerInfo -> {
- try {
- out.writeObject(workerInfo);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
-
- out.writeLong(partitionTotalWritten.sum());
- out.writeLong(partitionTotalFileCount.sum());
- out.writeObject(appDiskUsageMetric.snapShots());
- out.writeObject(appDiskUsageMetric.currentSnapShot());
-
- out.flush();
- }
-
- private <T> void writeSetMetaToFile(Set<T> metas, ObjectOutputStream out) throws IOException {
- out.writeInt(metas.size());
- metas.forEach(
- meta -> {
- try {
- out.writeObject(meta);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
+ byte[] snapshotBytes =
+ PbSerDeUtils.toPbSnapshotMetaInfo(
+ estimatedPartitionSize,
+ registeredShuffle,
+ hostnameSet,
+ blacklist,
+ workerLostEvents,
+ appHeartbeatTime,
+ workers,
+ partitionTotalWritten.sum(),
+ partitionTotalFileCount.sum(),
+ appDiskUsageMetric.snapShots(),
+ appDiskUsageMetric.currentSnapShot().get())
+ .toByteArray();
+ Files.write(file.toPath(), snapshotBytes);
}
/**
@@ -309,26 +268,23 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
* @throws IOException
*/
public void restoreMetaFromFile(File file) throws IOException {
- try (ObjectInputStream in =
- new ObjectInputStream(new BufferedInputStream(new FileInputStream(file)))) {
- estimatedPartitionSize = in.readLong();
- // read registeredShuffle
- readSetMetaFromFile(registeredShuffle, in.readInt(), in);
+ try (BufferedInputStream in = new BufferedInputStream(new FileInputStream(file))) {
+ PbSnapshotMetaInfo snapshotMetaInfo = PbSnapshotMetaInfo.parseFrom(in);
+
+ estimatedPartitionSize = snapshotMetaInfo.getEstimatedPartitionSize();
+
+ registeredShuffle.addAll(snapshotMetaInfo.getRegisteredShuffleList());
+ hostnameSet.addAll(snapshotMetaInfo.getHostnameSetList());
+ blacklist.addAll(
+ snapshotMetaInfo.getBlacklistList().stream()
+ .map(PbSerDeUtils::fromPbWorkerInfo)
+ .collect(Collectors.toSet()));
+ workerLostEvents.addAll(
+ snapshotMetaInfo.getWorkerLostEventsList().stream()
+ .map(PbSerDeUtils::fromPbWorkerInfo)
+ .collect(Collectors.toSet()));
+ appHeartbeatTime.putAll(snapshotMetaInfo.getAppHeartbeatTimeMap());
- // read hostnameSet
- readSetMetaFromFile(hostnameSet, in.readInt(), in);
-
- // read blacklist
- readSetMetaFromFile(blacklist, in.readInt(), in);
-
- // read workerLost events
- readSetMetaFromFile(workerLostEvents, in.readInt(), in);
-
- // read application heartbeat time
- int size = in.readInt();
- for (int i = 0; i < size; ++i) {
- appHeartbeatTime.put((String) in.readObject(), in.readLong());
- }
registeredShuffle.forEach(
shuffleKey -> {
String appId = shuffleKey.split("-")[0];
@@ -337,19 +293,24 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
}
});
- // read workerInfo
- int workersSize = in.readInt();
- for (int i = 0; i < workersSize; ++i) {
- workers.add((WorkerInfo) in.readObject());
- }
+ workers.addAll(
+ snapshotMetaInfo.getWorkersList().stream()
+ .map(PbSerDeUtils::fromPbWorkerInfo)
+ .collect(Collectors.toSet()));
+
partitionTotalWritten.reset();
- partitionTotalWritten.add(in.readLong());
+ partitionTotalWritten.add(snapshotMetaInfo.getPartitionTotalWritten());
partitionTotalFileCount.reset();
- partitionTotalFileCount.add(in.readLong());
- appDiskUsageMetric.restoreFromSnapshot((AppDiskUsageSnapShot[]) in.readObject());
+ partitionTotalFileCount.add(snapshotMetaInfo.getPartitionTotalFileCount());
+ appDiskUsageMetric.restoreFromSnapshot(
+ snapshotMetaInfo.getAppDiskUsageMetricSnapshotsList().stream()
+ .map(PbSerDeUtils::fromPbAppDiskUsageSnapshot)
+ .toArray(AppDiskUsageSnapShot[]::new));
appDiskUsageMetric.currentSnapShot_$eq(
- (AtomicReference<AppDiskUsageSnapShot>) in.readObject());
- } catch (ClassNotFoundException e) {
+ new AtomicReference<AppDiskUsageSnapShot>(
+ PbSerDeUtils.fromPbAppDiskUsageSnapshot(
+ snapshotMetaInfo.getCurrentAppDiskUsageMetricsSnapshot())));
+ } catch (Exception e) {
throw new IOException(e);
}
LOG.info("Successfully restore meta info from snapshot {}", file.getAbsolutePath());
@@ -362,13 +323,6 @@ public abstract class AbstractMetaManager implements IMetadataHandler {
registeredShuffle.forEach(shuffle -> LOG.info("RegisteredShuffle {}", shuffle));
}
- private <T> void readSetMetaFromFile(Set<T> metas, int size, ObjectInputStream in)
- throws ClassNotFoundException, IOException {
- for (int i = 0; i < size; ++i) {
- metas.add((T) in.readObject());
- }
- }
-
public void updateBlacklistByReportWorkerUnavailable(List<WorkerInfo> failedWorkers) {
synchronized (this.workers) {
failedWorkers.retainAll(this.workers);
diff --git a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
index 0ca95e40..afc4f009 100644
--- a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
+++ b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
@@ -39,6 +39,7 @@ import org.junit.Test;
import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.AppDiskUsageSnapShot;
import org.apache.celeborn.common.meta.DiskInfo;
import org.apache.celeborn.common.meta.WorkerInfo;
import org.apache.celeborn.common.quota.ResourceConsumption;
@@ -200,8 +201,9 @@ public class MasterStateMachineSuiteJ extends RatisBaseSuiteJ {
}
@Test
- public void testObjSerde() throws IOException {
- HAMasterMetaManager masterStatusSystem = new HAMasterMetaManager(null, new CelebornConf());
+ public void testObjSerde() throws IOException, InterruptedException {
+ CelebornConf conf = new CelebornConf();
+ HAMasterMetaManager masterStatusSystem = new HAMasterMetaManager(null, conf);
File tmpFile = File.createTempFile("tef", "test" + System.currentTimeMillis());
Map<String, DiskInfo> disks1 = new HashMap<>();
@@ -256,6 +258,22 @@ public class MasterStateMachineSuiteJ extends RatisBaseSuiteJ {
masterStatusSystem.hostnameSet.add(host2);
masterStatusSystem.hostnameSet.add(host3);
+ // Wait for update snapshot
+ Thread.sleep(60000);
+ Map<String, Long> appDiskUsage = new ConcurrentHashMap<String, Long>();
+ appDiskUsage.put("app-1", 100L);
+ appDiskUsage.put("app-2", 200L);
+ masterStatusSystem.appDiskUsageMetric.update(appDiskUsage);
+ appDiskUsage.put("app-3", 300L);
+ appDiskUsage.put("app-1", 200L);
+ masterStatusSystem.appDiskUsageMetric.update(appDiskUsage);
+ // wait for snapshot updated
+ Thread.sleep(3000);
+
+ AppDiskUsageSnapShot[] originSnapshots = masterStatusSystem.appDiskUsageMetric.snapShots();
+ AppDiskUsageSnapShot originCurrentSnapshot =
+ masterStatusSystem.appDiskUsageMetric.currentSnapShot().get();
+
masterStatusSystem.writeMetaInfoToFile(tmpFile);
masterStatusSystem.hostnameSet.clear();
@@ -265,5 +283,14 @@ public class MasterStateMachineSuiteJ extends RatisBaseSuiteJ {
Assert.assertEquals(3, masterStatusSystem.blacklist.size());
Assert.assertEquals(3, masterStatusSystem.hostnameSet.size());
+ Assert.assertEquals(
+ conf.metricsAppTopDiskUsageWindowSize(),
+ masterStatusSystem.appDiskUsageMetric.snapShots().length);
+ Assert.assertEquals(
+ conf.metricsAppTopDiskUsageCount(),
+ masterStatusSystem.appDiskUsageMetric.currentSnapShot().get().topNItems().length);
+ Assert.assertEquals(
+ originCurrentSnapshot, masterStatusSystem.appDiskUsageMetric.currentSnapShot().get());
+ Assert.assertEquals(originSnapshots, masterStatusSystem.appDiskUsageMetric.snapShots());
}
}