Source code for pyflink.datastream.slot_sharing_group
################################################################################
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
__all__ = ['MemorySize', 'SlotSharingGroup']
from typing import Optional
from pyflink.java_gateway import get_gateway
[docs]class MemorySize(object):
"""
MemorySize is a representation of a number of bytes, viewable in different units.
"""
def __init__(self, j_memory_size=None, bytes_size: int = None):
self._j_memory_size = get_gateway().jvm \
.org.apache.flink.configuration.MemorySize(bytes_size) \
if j_memory_size is None else j_memory_size
@staticmethod
def of_mebi_bytes(mebi_bytes: int) -> 'MemorySize':
return MemorySize(
get_gateway().jvm.org.apache.flink.configuration.MemorySize.ofMebiBytes(mebi_bytes))
def get_bytes(self) -> int:
"""
Gets the memory size in bytes.
:return: The memory size in bytes.
"""
return self._j_memory_size.getBytes()
def get_kibi_bytes(self) -> int:
"""
Gets the memory size in Kibibytes (= 1024 bytes).
:return: The memory size in Kibibytes.
"""
return self._j_memory_size.getKibiBytes()
def get_mebi_bytes(self) -> int:
"""
Gets the memory size in Mebibytes (= 1024 Kibibytes).
:return: The memory size in Mebibytes.
"""
return self._j_memory_size.getMebiBytes()
def get_gibi_bytes(self) -> int:
"""
Gets the memory size in Gibibytes (= 1024 Mebibytes).
:return: The memory size in Gibibytes.
"""
return self._j_memory_size.getGibiBytes()
def get_tebi_bytes(self) -> int:
"""
Gets the memory size in Tebibytes (= 1024 Gibibytes).
:return: The memory size in Tebibytes.
"""
return self._j_memory_size.getTebiBytes()
def get_java_memory_size(self):
"""
Gets the Java MemorySize object.
:return: The Java MemorySize object.
"""
return self._j_memory_size
def __eq__(self, other):
return isinstance(other, self.__class__) and self._j_memory_size == other._j_memory_size
def __hash__(self):
return self._j_memory_size.hashCode()
def __lt__(self, other: 'MemorySize'):
if not isinstance(other, MemorySize):
raise Exception("Does not support comparison with non-MemorySize %s" % other)
return self._j_memory_size.compareTo(other._j_memory_size) == -1
def __le__(self, other: 'MemorySize'):
return self.__eq__(other) and self.__lt__(other)
def __str__(self):
return self._j_memory_size.toString()
[docs]class SlotSharingGroup(object):
"""
Describe the name and the different resource components of a slot sharing group.
"""
def __init__(self, j_slot_sharing_group):
self._j_slot_sharing_group = j_slot_sharing_group
def get_name(self) -> str:
"""
Gets the name of this SlotSharingGroup.
:return: The name of the SlotSharingGroup.
"""
return self._j_slot_sharing_group.getName()
def get_managed_memory(self) -> Optional[MemorySize]:
"""
Gets the task managed memory for this SlotSharingGroup.
:return: The task managed memory of the SlotSharingGroup.
"""
managed_memory = self._j_slot_sharing_group.getManagedMemory()
return MemorySize(managed_memory.get()) if managed_memory.isPresent() else None
def get_task_heap_memory(self) -> Optional[MemorySize]:
"""
Gets the task heap memory for this SlotSharingGroup.
:return: The task heap memory of the SlotSharingGroup.
"""
task_heap_memory = self._j_slot_sharing_group.getTaskHeapMemory()
return MemorySize(task_heap_memory.get()) if task_heap_memory.isPresent() else None
def get_task_off_heap_memory(self) -> Optional[MemorySize]:
"""
Gets the task off-heap memory for this SlotSharingGroup.
:return: The task off-heap memory of the SlotSharingGroup.
"""
task_off_heap_memory = self._j_slot_sharing_group.getTaskOffHeapMemory()
return MemorySize(task_off_heap_memory.get()) if task_off_heap_memory.isPresent() else None
def get_cpu_cores(self) -> Optional[float]:
"""
Gets the CPU cores for this SlotSharingGroup.
:return: The CPU cores of the SlotSharingGroup.
"""
cpu_cores = self._j_slot_sharing_group.getCpuCores()
return cpu_cores.get() if cpu_cores.isPresent() else None
def get_external_resources(self) -> dict:
"""
Gets the external resource from this SlotSharingGroup.
:return: User specified resources of the SlotSharingGroup.
"""
return dict(self._j_slot_sharing_group.getExternalResources())
def get_java_slot_sharing_group(self):
"""
Gets the Java SlotSharingGroup object.
:return: The Java SlotSharingGroup object.
"""
return self._j_slot_sharing_group
@staticmethod
def builder(name: str) -> 'Builder':
"""
Gets the Builder with the given name for this SlotSharingGroup.
:param name: The name of the SlotSharingGroup.
:return: The builder for the SlotSharingGroup.
"""
return SlotSharingGroup.Builder(
get_gateway().jvm.org.apache.flink.api.common.operators.SlotSharingGroup.newBuilder(
name))
def __eq__(self, other):
return isinstance(other, self.__class__) and \
self._j_slot_sharing_group == other._j_slot_sharing_group
def __hash__(self):
return self._j_slot_sharing_group.hashCode()
class Builder(object):
"""
Builder for the SlotSharingGroup.
"""
def __init__(self, j_builder):
self._j_builder = j_builder
def set_cpu_cores(self, cpu_cores: float) -> 'SlotSharingGroup.Builder':
"""
Sets the CPU cores for this SlotSharingGroup.
:param cpu_cores: The CPU cores of the SlotSharingGroup.
:return: This object.
"""
self._j_builder.setCpuCores(cpu_cores)
return self
def set_task_heap_memory(self, task_heap_memory: MemorySize) -> 'SlotSharingGroup.Builder':
"""
Sets the task heap memory for this SlotSharingGroup.
:param task_heap_memory: The task heap memory of the SlotSharingGroup.
:return: This object.
"""
self._j_builder.setTaskHeapMemory(task_heap_memory.get_java_memory_size())
return self
def set_task_heap_memory_mb(self, task_heap_memory_mb: int) -> 'SlotSharingGroup.Builder':
"""
Sets the task heap memory for this SlotSharingGroup in MB.
:param task_heap_memory_mb: The task heap memory of the SlotSharingGroup in MB.
:return: This object.
"""
self._j_builder.setTaskHeapMemoryMB(task_heap_memory_mb)
return self
def set_task_off_heap_memory(self, task_off_heap_memory: MemorySize) \
-> 'SlotSharingGroup.Builder':
"""
Sets the task off-heap memory for this SlotSharingGroup.
:param task_off_heap_memory: The task off-heap memory of the SlotSharingGroup.
:return: This object.
"""
self._j_builder.setTaskOffHeapMemory(task_off_heap_memory.get_java_memory_size())
return self
def set_task_off_heap_memory_mb(self, task_off_heap_memory_mb: int) \
-> 'SlotSharingGroup.Builder':
"""
Sets the task off-heap memory for this SlotSharingGroup in MB.
:param task_off_heap_memory_mb: The task off-heap memory of the SlotSharingGroup in MB.
:return: This object.
"""
self._j_builder.setTaskOffHeapMemoryMB(task_off_heap_memory_mb)
return self
def set_managed_memory(self, managed_memory: MemorySize) -> 'SlotSharingGroup.Builder':
"""
Sets the task managed memory for this SlotSharingGroup.
:param managed_memory: The task managed memory of the SlotSharingGroup.
:return: This object.
"""
self._j_builder.setManagedMemory(managed_memory.get_java_memory_size())
return self
def set_managed_memory_mb(self, managed_memory_mb: int) -> 'SlotSharingGroup.Builder':
"""
Sets the task managed memory for this SlotSharingGroup in MB.
:param managed_memory_mb: The task managed memory of the SlotSharingGroup in MB.
:return: This object.
"""
self._j_builder.setManagedMemoryMB(managed_memory_mb)
return self
def set_external_resource(self, name: str, value: float) -> 'SlotSharingGroup.Builder':
"""
Adds the given external resource. The old value with the same resource name will be
replaced if present.
:param name: The resource name of the given external resource.
:param value: The value of the given external resource.
:return: This object.
"""
self._j_builder.setExternalResource(name, value)
return self
def build(self) -> 'SlotSharingGroup':
"""
Builds the SlotSharingGroup.
:return: The SlotSharingGroup object.
"""
return SlotSharingGroup(j_slot_sharing_group=self._j_builder.build())