共计 14044 个字符，预计需要花费 36 分钟才能阅读完成。
序
本文主要研究一下 flink 的 BlobService
BlobService
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobService.java
/**
* A simple store and retrieve binary large objects (BLOBs).
*/
public interface BlobService extends Closeable {
/**
* Returns a BLOB service for accessing permanent BLOBs.
*
* @return BLOB service
*/
PermanentBlobService getPermanentBlobService();
/**
* Returns a BLOB service for accessing transient BLOBs.
*
* @return BLOB service
*/
TransientBlobService getTransientBlobService();
/**
* Returns the port of the BLOB server that this BLOB service is working with.
*
* @return the port the blob server.
*/
int getPort();
}
BlobService 定义了 getPermanentBlobService 方法用于获取 PermanentBlobService；getTransientBlobService 方法用于获取 TransientBlobService
PermanentBlobService
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobService.java
/**
* A service to retrieve permanent binary large objects (BLOBs).
*
* <p>These may include per-job BLOBs that are covered by high-availability (HA) mode, e.g. a job’s
* JAR files or (parts of) an off-loaded {@link org.apache.flink.runtime.deployment.TaskDeploymentDescriptor}
* or files in the {@link org.apache.flink.api.common.cache.DistributedCache}.
*/
public interface PermanentBlobService extends Closeable {
/**
* Returns the path to a local copy of the file associated with the provided job ID and blob
* key.
*
* @param jobId
* ID of the job this blob belongs to
* @param key
* BLOB key associated with the requested file
*
* @return The path to the file.
*
* @throws java.io.FileNotFoundException
* if the BLOB does not exist;
* @throws IOException
* if any other error occurs when retrieving the file
*/
File getFile(JobID jobId, PermanentBlobKey key) throws IOException;
}
PermanentBlobService 提供了 getFile 方法，它根据 JobID 及 PermanentBlobKey 来获取 File
TransientBlobService
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobService.java
/**
* A service to retrieve transient binary large objects (BLOBs) which are deleted on the
* {@link BlobServer} when they are retrieved.
*
* <p>These may include per-job BLOBs like files in the {@link
* org.apache.flink.api.common.cache.DistributedCache}, for example.
*
* <p>Note: None of these BLOBs is highly available (HA). This case is covered by BLOBs in the
* {@link PermanentBlobService}.
*
* <p>TODO: change API to not rely on local files but return {@link InputStream} objects
*/
public interface TransientBlobService extends Closeable {
// ——————————————————————————————–
// GET
// ——————————————————————————————–
/**
* Returns the path to a local copy of the (job-unrelated) file associated with the provided
* blob key.
*
* @param key
* blob key associated with the requested file
*
* @return The path to the file.
*
* @throws java.io.FileNotFoundException
* when the path does not exist;
* @throws IOException
* if any other error occurs when retrieving the file
*/
File getFile(TransientBlobKey key) throws IOException;
/**
* Returns the path to a local copy of the file associated with the provided job ID and blob
* key.
*
* @param jobId
* ID of the job this blob belongs to
* @param key
* blob key associated with the requested file
*
* @return The path to the file.
*
* @throws java.io.FileNotFoundException
* when the path does not exist;
* @throws IOException
* if any other error occurs when retrieving the file
*/
File getFile(JobID jobId, TransientBlobKey key) throws IOException;
// ——————————————————————————————–
// PUT
// ——————————————————————————————–
/**
* Uploads the (job-unrelated) data of the given byte array to the BLOB server.
*
* @param value
* the buffer to upload
*
* @return the computed BLOB key identifying the BLOB on the server
*
* @throws IOException
* thrown if an I/O error occurs while uploading the data to the BLOB server
*/
TransientBlobKey putTransient(byte[] value) throws IOException;
/**
* Uploads the data of the given byte array for the given job to the BLOB server.
*
* @param jobId
* the ID of the job the BLOB belongs to
* @param value
* the buffer to upload
*
* @return the computed BLOB key identifying the BLOB on the server
*
* @throws IOException
* thrown if an I/O error occurs while uploading the data to the BLOB server
*/
TransientBlobKey putTransient(JobID jobId, byte[] value) throws IOException;
/**
* Uploads the (job-unrelated) data from the given input stream to the BLOB server.
*
* @param inputStream
* the input stream to read the data from
*
* @return the computed BLOB key identifying the BLOB on the server
*
* @throws IOException
* thrown if an I/O error occurs while reading the data from the input stream or uploading the
* data to the BLOB server
*/
TransientBlobKey putTransient(InputStream inputStream) throws IOException;
/**
* Uploads the data from the given input stream for the given job to the BLOB server.
*
* @param jobId
* ID of the job this blob belongs to
* @param inputStream
* the input stream to read the data from
*
* @return the computed BLOB key identifying the BLOB on the server
*
* @throws IOException
* thrown if an I/O error occurs while reading the data from the input stream or uploading the
* data to the BLOB server
*/
TransientBlobKey putTransient(JobID jobId, InputStream inputStream) throws IOException;
// ——————————————————————————————–
// DELETE
// ——————————————————————————————–
/**
* Deletes the (job-unrelated) file associated with the provided blob key from the local cache.
*
* @param key
* associated with the file to be deleted
*
* @return <tt>true</tt> if the given blob is successfully deleted or non-existing;
* <tt>false</tt> otherwise
*/
boolean deleteFromCache(TransientBlobKey key);
/**
* Deletes the file associated with the provided job ID and blob key from the local cache.
*
* @param jobId
* ID of the job this blob belongs to
* @param key
* associated with the file to be deleted
*
* @return <tt>true</tt> if the given blob is successfully deleted or non-existing;
* <tt>false</tt> otherwise
*/
boolean deleteFromCache(JobID jobId, TransientBlobKey key);
}
TransientBlobService 用于获取 transient binary large objects (BLOBs)，这些 blobs 在获取时就会在 BlobServer 上删掉；它提供了 getFile、putTransient、deleteFromCache 方法
BlobKey
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
/**
* A BLOB key uniquely identifies a BLOB.
*/
public abstract class BlobKey implements Serializable, Comparable<BlobKey> {
private static final long serialVersionUID = 3847117712521785209L;
/** Size of the internal BLOB key in bytes. */
public static final int SIZE = 20;
/** The byte buffer storing the actual key data. */
private final byte[] key;
/**
* (Internal) BLOB type – to be reflected by the inheriting sub-class.
*/
private final BlobType type;
/**
* BLOB type, i.e. permanent or transient.
*/
enum BlobType {
/**
* Indicates a permanent BLOB whose lifecycle is that of a job and which is made highly
* available.
*/
PERMANENT_BLOB,
/**
* Indicates a transient BLOB whose lifecycle is managed by the user and which is not made
* highly available.
*/
TRANSIENT_BLOB
}
/**
* Random component of the key.
*/
private final AbstractID random;
/**
* Constructs a new BLOB key.
*
* @param type
* whether the referenced BLOB is permanent or transient
*/
protected BlobKey(BlobType type) {
this.type = checkNotNull(type);
this.key = new byte[SIZE];
this.random = new AbstractID();
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param type
* whether the referenced BLOB is permanent or transient
* @param key
* the actual key data
*/
protected BlobKey(BlobType type, byte[] key) {
if (key == null || key.length != SIZE) {
throw new IllegalArgumentException(“BLOB key must have a size of ” + SIZE + ” bytes”);
}
this.type = checkNotNull(type);
this.key = key;
this.random = new AbstractID();
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param type
* whether the referenced BLOB is permanent or transient
* @param key
* the actual key data
* @param random
* the random component of the key
*/
protected BlobKey(BlobType type, byte[] key, byte[] random) {
if (key == null || key.length != SIZE) {
throw new IllegalArgumentException(“BLOB key must have a size of ” + SIZE + ” bytes”);
}
this.type = checkNotNull(type);
this.key = key;
this.random = new AbstractID(random);
}
/**
* Returns the right {@link BlobKey} subclass for the given parameters.
*
* @param type
* whether the referenced BLOB is permanent or transient
*
* @return BlobKey subclass
*/
@VisibleForTesting
static BlobKey createKey(BlobType type) {
if (type == PERMANENT_BLOB) {
return new PermanentBlobKey();
} else {
return new TransientBlobKey();
}
}
/**
* Returns the right {@link BlobKey} subclass for the given parameters.
*
* @param type
* whether the referenced BLOB is permanent or transient
* @param key
* the actual key data
*
* @return BlobKey subclass
*/
static BlobKey createKey(BlobType type, byte[] key) {
if (type == PERMANENT_BLOB) {
return new PermanentBlobKey(key);
} else {
return new TransientBlobKey(key);
}
}
/**
* Returns the right {@link BlobKey} subclass for the given parameters.
*
* @param type
* whether the referenced BLOB is permanent or transient
* @param key
* the actual key data
* @param random
* the random component of the key
*
* @return BlobKey subclass
*/
static BlobKey createKey(BlobType type, byte[] key, byte[] random) {
if (type == PERMANENT_BLOB) {
return new PermanentBlobKey(key, random);
} else {
return new TransientBlobKey(key, random);
}
}
/**
* Returns the hash component of this key.
*
* @return a 20 bit hash of the contents the key refers to
*/
@VisibleForTesting
public byte[] getHash() {
return key;
}
/**
* Returns the (internal) BLOB type which is reflected by the inheriting sub-class.
*
* @return BLOB type, i.e. permanent or transient
*/
BlobType getType() {
return type;
}
/**
* Adds the BLOB key to the given {@link MessageDigest}.
*
* @param md
* the message digest to add the BLOB key to
*/
public void addToMessageDigest(MessageDigest md) {
md.update(this.key);
}
@Override
public boolean equals(final Object obj) {
if (!(obj instanceof BlobKey)) {
return false;
}
final BlobKey bk = (BlobKey) obj;
return Arrays.equals(this.key, bk.key) &&
this.type == bk.type &&
this.random.equals(bk.random);
}
@Override
public int hashCode() {
int result = Arrays.hashCode(this.key);
result = 37 * result + this.type.hashCode();
result = 37 * result + this.random.hashCode();
return result;
}
@Override
public String toString() {
final String typeString;
switch (this.type) {
case TRANSIENT_BLOB:
typeString = “t-“;
break;
case PERMANENT_BLOB:
typeString = “p-“;
break;
default:
// this actually never happens!
throw new IllegalStateException(“Invalid BLOB type”);
}
return typeString + StringUtils.byteToHexString(this.key) + “-” + random.toString();
}
@Override
public int compareTo(BlobKey o) {
// compare the hashes first
final byte[] aarr = this.key;
final byte[] barr = o.key;
final int len = Math.min(aarr.length, barr.length);
for (int i = 0; i < len; ++i) {
final int a = (aarr[i] & 0xff);
final int b = (barr[i] & 0xff);
if (a != b) {
return a – b;
}
}
if (aarr.length == barr.length) {
// same hash contents – compare the BLOB types
int typeCompare = this.type.compareTo(o.type);
if (typeCompare == 0) {
// same type – compare random components
return this.random.compareTo(o.random);
} else {
return typeCompare;
}
} else {
return aarr.length – barr.length;
}
}
// ——————————————————————————————–
/**
* Auxiliary method to read a BLOB key from an input stream.
*
* @param inputStream
* the input stream to read the BLOB key from
* @return the read BLOB key
* @throws IOException
* throw if an I/O error occurs while reading from the input stream
*/
static BlobKey readFromInputStream(InputStream inputStream) throws IOException {
final byte[] key = new byte[BlobKey.SIZE];
final byte[] random = new byte[AbstractID.SIZE];
int bytesRead = 0;
// read key
while (bytesRead < key.length) {
final int read = inputStream.read(key, bytesRead, key.length – bytesRead);
if (read < 0) {
throw new EOFException(“Read an incomplete BLOB key”);
}
bytesRead += read;
}
// read BLOB type
final BlobType blobType;
{
final int read = inputStream.read();
if (read < 0) {
throw new EOFException(“Read an incomplete BLOB type”);
} else if (read == TRANSIENT_BLOB.ordinal()) {
blobType = TRANSIENT_BLOB;
} else if (read == PERMANENT_BLOB.ordinal()) {
blobType = PERMANENT_BLOB;
} else {
throw new IOException(“Invalid data received for the BLOB type: ” + read);
}
}
// read random component
bytesRead = 0;
while (bytesRead < AbstractID.SIZE) {
final int read = inputStream.read(random, bytesRead, AbstractID.SIZE – bytesRead);
if (read < 0) {
throw new EOFException(“Read an incomplete BLOB key”);
}
bytesRead += read;
}
return createKey(blobType, key, random);
}
/**
* Auxiliary method to write this BLOB key to an output stream.
*
* @param outputStream
* the output stream to write the BLOB key to
* @throws IOException
* thrown if an I/O error occurs while writing the BLOB key
*/
void writeToOutputStream(final OutputStream outputStream) throws IOException {
outputStream.write(this.key);
outputStream.write(this.type.ordinal());
outputStream.write(this.random.getBytes());
}
}
BlobKey 是个抽象类，它有 key、BlobType、AbstractID 三个属性，其中 BlobType 分为 PERMANENT_BLOB 及 TRANSIENT_BLOB；它定义了 createKey 静态方法，用于根据 BlobType 创建 BlobKey；readFromInputStream 方法用于从 InputStream 反序列化为 BlobKey；writeToOutputStream 方法用于将 BlobKey 序列化到 OutputStream；它有两个子类，分别为 PermanentBlobKey 及 TransientBlobKey
PermanentBlobKey
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/PermanentBlobKey.java
/**
* BLOB key referencing permanent BLOB files.
*/
public final class PermanentBlobKey extends BlobKey {
/**
* Constructs a new BLOB key.
*/
@VisibleForTesting
public PermanentBlobKey() {
super(BlobType.PERMANENT_BLOB);
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param key
* the actual key data
*/
PermanentBlobKey(byte[] key) {
super(BlobType.PERMANENT_BLOB, key);
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param key
* the actual key data
* @param random
* the random component of the key
*/
PermanentBlobKey(byte[] key, byte[] random) {
super(BlobType.PERMANENT_BLOB, key, random);
}
}
PermanentBlobKey 继承了 BlobKey，它的 BlobType 为 BlobType.PERMANENT_BLOB
TransientBlobKey
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/blob/TransientBlobKey.java
/**
* BLOB key referencing transient BLOB files.
*/
public final class TransientBlobKey extends BlobKey {
/**
* Constructs a new BLOB key.
*/
@VisibleForTesting
public TransientBlobKey() {
super(BlobType.TRANSIENT_BLOB);
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param key
* the actual key data
*/
TransientBlobKey(byte[] key) {
super(BlobType.TRANSIENT_BLOB, key);
}
/**
* Constructs a new BLOB key from the given byte array.
*
* @param key
* the actual key data
* @param random
* the random component of the key
*/
TransientBlobKey(byte[] key, byte[] random) {
super(BlobType.TRANSIENT_BLOB, key, random);
}
}
TransientBlobKey 继承了 BlobKey，它的 BlobType 为 BlobType.TRANSIENT_BLOB
小结
BlobService 定义了 getPermanentBlobService 方法用于获取 PermanentBlobService；getTransientBlobService 方法用于获取 TransientBlobService
PermanentBlobService 提供了 getFile 方法，它根据 JobID 及 PermanentBlobKey 来获取 File；TransientBlobService 用于获取 transient binary large objects (BLOBs)，这些 blobs 在获取时就会在 BlobServer 上删掉；它提供了 getFile、putTransient、deleteFromCache 方法
BlobKey 是个抽象类，它有 key、BlobType、AbstractID 三个属性，其中 BlobType 分为 PERMANENT_BLOB 及 TRANSIENT_BLOB；它定义了 createKey 静态方法，用于根据 BlobType 创建 BlobKey；readFromInputStream 方法用于从 InputStream 反序列化为 BlobKey；writeToOutputStream 方法用于将 BlobKey 序列化到 OutputStream；它有两个子类，分别为 PermanentBlobKey 及 TransientBlobKey；PermanentBlobKey 继承了 BlobKey，它的 BlobType 为 BlobType.PERMANENT_BLOB；TransientBlobKey 继承了 BlobKey，它的 BlobType 为 BlobType.TRANSIENT_BLOB
doc
Blob Server
