org.apache.samza.table.descriptors

Class LocalTableDescriptor<K,V,D extends LocalTableDescriptor<K,V,D>>

java.lang.Object

org.apache.samza.table.descriptors.BaseTableDescriptor<K,V,D>

org.apache.samza.table.descriptors.LocalTableDescriptor<K,V,D>

Type Parameters:

K - the type of the key in this table

V - the type of the value in this table

D - the type of the concrete table descriptor

All Implemented Interfaces:

TableDescriptor<K,V,D>

public abstract class LocalTableDescriptor<K,V,D extends LocalTableDescriptor<K,V,D>>
extends BaseTableDescriptor<K,V,D>

Table descriptor for store backed tables.

Field Summary

Fields

Modifier and Type	Field and Description
protected java.lang.Integer	changelogReplicationFactor
protected java.lang.String	changelogStream
protected boolean	enableChangelog
protected KVSerde<K,V>	serde
protected java.util.List<java.lang.String>	sideInputs
protected SideInputsProcessor	sideInputsProcessor
static java.util.regex.Pattern	SYSTEM_STREAM_NAME_PATTERN

Fields inherited from class org.apache.samza.table.descriptors.BaseTableDescriptor

config, tableId

Constructor Summary

Constructors

Constructor and Description
LocalTableDescriptor(java.lang.String tableId, KVSerde<K,V> serde) Constructs a table descriptor instance

Method Summary

Modifier and Type	Method and Description
protected void	addStoreConfig(java.lang.String key, java.lang.String value, java.util.Map<java.lang.String,java.lang.String> tableConfig) Helper method to add a store level config item to table configuration
KVSerde<K,V>	getSerde() Get the serde assigned to this TableDescriptor
java.util.List<java.lang.String>	getSideInputs() Get side input stream names
java.util.Map<java.lang.String,java.lang.String>	toConfig(Config jobConfig) Generate configuration for this table descriptor, the generated configuration should be the complete configuration for this table that can be directly included in the job configuration.
protected void	validate() Validate that this table descriptor is constructed properly; this method is used internally.
D	withChangelogEnabled() Enable changelog for this table, by default changelog is disabled.
D	withChangelogReplicationFactor(int replicationFactor) The property defines the number of replicas to use for the change log stream.
D	withChangelogStream(java.lang.String changelogStream) Samza stores are local to a container.
D	withSideInputs(java.util.List<java.lang.String> sideInputs) Add side inputs to the table.
D	withSideInputsProcessor(SideInputsProcessor sideInputsProcessor) Provide the SideInputsProcessor for this table.

Methods inherited from class org.apache.samza.table.descriptors.BaseTableDescriptor

addTableConfig, getProviderFactoryClassName, getTableId, withConfig

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

SYSTEM_STREAM_NAME_PATTERN

public static final java.util.regex.Pattern SYSTEM_STREAM_NAME_PATTERN

serde

protected final KVSerde<K,V> serde

enableChangelog

protected boolean enableChangelog

changelogStream

protected java.lang.String changelogStream

changelogReplicationFactor

protected java.lang.Integer changelogReplicationFactor

sideInputs

protected java.util.List<java.lang.String> sideInputs

sideInputsProcessor

protected SideInputsProcessor sideInputsProcessor

Constructor Detail

LocalTableDescriptor

public LocalTableDescriptor(java.lang.String tableId,
                            KVSerde<K,V> serde)

Constructs a table descriptor instance

Parameters:

tableId - Id of the table, it must conform to pattern [\\d\\w-_]+

serde - the serde for key and value

Method Detail

withSideInputs

public D withSideInputs(java.util.List<java.lang.String> sideInputs)

Add side inputs to the table. Each stream is of the format system-name.stream-name. The streams are marked as bootstrap streams and once the table is bootstrapped, it is updated in the background in change capture mode. Applications should specify the transformation logic using withSideInputsProcessor(SideInputsProcessor), which is will be applied to the incoming messages and the results are written to the table.

Parameters:

sideInputs - list of side input streams

Returns:

this table descriptor instance

withSideInputsProcessor

public D withSideInputsProcessor(SideInputsProcessor sideInputsProcessor)

Provide the SideInputsProcessor for this table. It is applied on the side inputs and the results are written to the table.

Parameters:

sideInputsProcessor - a side input processor

Returns:

this table descriptor instance

withChangelogEnabled

public D withChangelogEnabled()

Enable changelog for this table, by default changelog is disabled. When the changelog stream name is not specified, it is automatically generated in the format [job-name]-[job-id]-table-[table-id]. Refer to stores.store-name.changelog in Samza configuration guide

Returns:

this table descriptor instance

withChangelogStream

public D withChangelogStream(java.lang.String changelogStream)

Samza stores are local to a container. If the container fails, the contents of the store are lost. To prevent loss of data, you need to set this property to configure a changelog stream: Samza then ensures that writes to the store are replicated to this stream, and the store is restored from this stream after a failure. The value of this property is given in the form system-name.stream-name. The "system-name" part is optional. If it is omitted you must specify the system in job.changelog.system config. Any output stream can be used as changelog, but you must ensure that only one job ever writes to a given changelog stream (each instance of a job and each store needs its own changelog stream).

Refer to stores.store-name.changelog in Samza configuration guide

Parameters:

changelogStream - changelog stream name

Returns:

this table descriptor instance

withChangelogReplicationFactor

public D withChangelogReplicationFactor(int replicationFactor)

The property defines the number of replicas to use for the change log stream.

Default value is stores.default.changelog.replication.factor.

Refer to stores.store-name.changelog.replication.factor in Samza configuration guide

Parameters:

replicationFactor - replication factor

Returns:

this table descriptor instance

toConfig

public java.util.Map<java.lang.String,java.lang.String> toConfig(Config jobConfig)

Generate configuration for this table descriptor, the generated configuration should be the complete configuration for this table that can be directly included in the job configuration. Note: although the serdes may have already been set in this instance, their corresponding configuration needs to be generated centrally for consistency and efficiency reasons. Therefore the serde configuration for this table is expected to have already been generated and stored in the jobConfig. Note: Serdes are expected to be generated during configuration generation of the job node, which is not handled here.

Specified by:

toConfig in interface TableDescriptor<K,V,D extends LocalTableDescriptor<K,V,D>>

Overrides:

toConfig in class BaseTableDescriptor<K,V,D extends LocalTableDescriptor<K,V,D>>

Parameters:

jobConfig - job configuration

Returns:

table configuration

getSideInputs

public java.util.List<java.lang.String> getSideInputs()

Get side input stream names

Returns:

side inputs

getSerde

public KVSerde<K,V> getSerde()

Get the serde assigned to this TableDescriptor

Returns:

KVSerde used by this table

validate

protected void validate()

Description copied from class: BaseTableDescriptor

Validate that this table descriptor is constructed properly; this method is used internally.

Specified by:

validate in class BaseTableDescriptor<K,V,D extends LocalTableDescriptor<K,V,D>>

addStoreConfig

protected void addStoreConfig(java.lang.String key,
                              java.lang.String value,
                              java.util.Map<java.lang.String,java.lang.String> tableConfig)

Helper method to add a store level config item to table configuration

Parameters:

key - key of the config item

value - value of the config item

tableConfig - table configuration