...
The Application now shows up in all the normal RESTful APIs, with all its programs, streams, and datasets.
1.3 Updating an Application
...
| Code Block |
|---|
POST /namespaces/default/artifacts/myapp --data-binary @myapp-1.0.1.jar |
Note: Artifacts are immutable unless they are snapshot versions. Deploying again to version 1.0.0 would cause a conflict error.
A call can be made to determine if there are any Applications using the older artifact:
...
2. Create an Application that uses plugins
2.1
...
Application Class changes
Now we decide the user decides to update our the MyApp Application Class to support pluggable ways of reading from a stream. We do this This is done by introducing a 'ReaderStreamReader' interface :
| Code Block |
|---|
|
...
in their project:
| Code Block |
|---|
public |
...
interface StreamReader { Put |
...
read(StreamEvent event);
} |
The user wants this StreamReader interface to be pluggable. There can be many implementations of StreamReader, and which implementation to use should be configurable. The Flowlet code changes to use the new StreamReader interface using the plugin java API:
| Code Block |
|---|
public class Reader extends AbstractFlowlet { @Property private String tableName; private Table table; private StreamReader streamReader; Reader(String tableName) { this.tableName = tableName; } @Override public void initialize(FlowletContext context) throws Exception { table = context.getDataset(tableName); streamReader = |
...
context.newPluginInstance(" |
...
readerPluginID"); } |
...
|
...
|
...
@ProcessInput public void |
...
process(StreamEvent event) { |
...
table.put(streamReader.read(event)); } } |
The Application Class is changed to register a "streamreader" plugin based on configuration:
| Code Block |
|---|
public class MyApp extends |
...
AbstractApplication<MyApp.MyConfig> { public static class MyConfig extends Config { |
...
@Nullable |
...
|
...
|
...
@Description("The name of |
...
the |
...
stream |
...
to |
...
read |
...
from. |
...
Defaults to 'A'.") |
...
|
...
|
...
|
...
|
...
private |
...
String |
...
stream; |
...
|
...
@Nullable @Description("The name of the |
...
table |
...
to |
...
write |
...
to. |
...
Defaults |
...
to 'X'.") |
...
|
...
private |
...
String table; |
...
@Description( |
...
"The name of the streamreader plugin to use.") |
...
private String readerPlugin; |
...
|
...
@Nullable |
...
|
...
|
...
|
...
@Description("Properties to send to the streamreader plugin.") @PluginType("streamreader") private |
...
PluginProperties |
...
readerPluginProperties; |
...
private |
...
MyConfig() { |
...
|
...
|
...
this.stream |
...
= "A"; |
...
this.table |
...
= |
...
"X"; |
...
} |
...
|
...
} |
...
@Override |
...
public void configure() { // |
...
ApplicationContext |
...
now |
...
has a method to get |
...
a |
...
custom |
...
config object whose fields will |
...
|
...
|
...
// |
...
be injected using the values |
...
given in the |
...
RESTful |
...
API |
...
|
...
MyConfig config |
...
= |
...
getContext().getConfig(); |
...
addStream(new Stream(config.stream)); |
...
createDataset(config.table, Table.class); addFlow(new MyFlow(config.stream, config.table, config.flowConfig)); // |
...
arguments are: type, name, id, properties |
...
usePlugin(" |
...
streamreader", |
...
config.readerPlugin, "readerPluginID", config.readerPluginProperties); |
...
1. Deploying an Artifact
A development team creates a project built on top of CDAP. Their CI build runs and produces a jar file. An administrator deploys the jar by making a REST call:
| Code Block |
|---|
POST /namespaces/default/artifacts/myapp --data-binary @myapp-1.0.0.jar |
CDAP opens the jar, figures out the artifact version based on the the bundle-version in the manifest, figures out what apps, programs, datasets, and plugins are in the artifact, then stores the artifact on the filesystem and metadata in a table.
The administrator can examine the metadata by making a call:
| Code Block |
|---|
GET /namespaces/default/artifacts/myapp/versions/1.0.0
{
"name": "myapp",
"version": "1.0.0",
"classes": {
"apps": [
{
"className": "co.cask.cdap.examples.myapp.MyApp",
"properties": {
"stream": {
"name": "stream",
"description": "The name of the stream to read from. Defaults to 'A'.",
"type": "string",
"required": false
},
"table": {
"name": "table",
"description": "The name of the table to write to. Defaults to 'X'.",
"type": "string",
"required": false,
},
"flowConfig": {
"name": "flow",
"description": "",
"type": "config",
"fields": {
"reader": {
"name": "reader",
"description": "",
"type": "config",
"required": true,
"fields": {
"name": {
"name": "name",
"description": "The name of the reader plugin to use.",
"type": "string",
"required": true
},
"properties": {
"name": "properties",
"description": "The properties needed by the chosen reader plugin.",
"type": "plugin",
"plugintype": "reader",
"required": true
}
}
},
"writer": { ... }
}
}
}
}
],
"plugins": [
{
"name": "default",
"type": "reader",
"description": "Writes timestamp and body as two columns and expects the row key to come as a header in the stream event.",
"className": "co.cask.cdap.examples.myapp.plugins.DefaultStreamReader",
"properties": {
"rowkey": {
"name": "rowkey",
"description": "The header that should be used as the row key to write to. Defaults to 'rowkey'.",
"type": "string",
"required": false
}
}
}
],
"flows": [ ... ],
"flowlets": [ ... ],
"datasetModules": [ ... ]
}
} |
Reverse indices will be maintained to allow querying the classes in artifacts directly:
| Code Block |
|---|
GET /namespaces/default/artifacts/myapp/versions/1.0.0/extensions
[ "reader" ]
GET /namespaces/default/artifacts/myapp/versions/1.0.0/extensions/reader
[
{
"type": "reader",
"name": "default",
"description": "Writes timestamp and body as two columns and expects the row key to come as a header in the stream event.",
"className": "co.cask.cdap.examples.myapp.plugins.DefaultStreamReader"
"extendsArtifact": {
"name": "myapp",
"versions": "[1.0.0, 2.0.0)"
}
"pluginArtifact": {
"name": "readers",
"version": "0.6.0"
}
}
]
GET /namespaces/default/artifacts/myapp/versions/1.0.0/extensions/reader/plugins/default
{
"name": "default",
"type": "reader",
"description": "Writes timestamp and body as two columns and expects the row key to come as a header in the stream event.",
"className": "co.cask.cdap.examples.myapp.plugins.DefaultStreamReader",
"properties": {
"rowkey": {
"name": "rowkey",
"description": "The header that should be used as the row key to write to. Defaults to 'rowkey'.",
"type": "string",
"required": false
}
}
}
GET /namespaces/default/appClasses
[
{
"className": "co.cask.cdap.examples.myapp.MyApp",
"description": "",
"artifact": {
"namespace": "default",
"name": "myapp",
"version": "1.0.0"
}
}
] |
2. Creating an Application
The administrator notices there is an app 'co.cask.cdap.examples.myapp.MyApp' contained in the artifact. Based on the app properties, the admin gathers that it needs a config of the form:
| Code Block |
|---|
{ "stream": "A", "table": "X", "flow": { "reader": { "name": "<some plugin name>", "properties": { <properties for plugins of type "reader"> }} } |
This becomes v2 of the Application Class. It is deployed via the same RESTful API:
| Code Block |
|---|
POST /namespaces/default/artifacts/myapp --data-binary @myapp-2.0.0.jar |
The metadata about this artifact now includes additional information about the config:
| Code Block |
|---|
GET /namespaces/default/artifacts/myapp/versions/2.0.0 { "name": "myapp", "version": "2.0.0", "classes": { "apps": [ { "className": "co.cask.cdap.examples.myapp.MyApp", "properties": { "stream": { "name": "stream", "description": "The name of the stream to read from. Defaults to 'A'.", "type": "string", "required": false }, "writer "table": { ... } } } |
He then makes a call to see what plugins of type 'reader' are available:
| Code Block |
|---|
GET /namespaces/default/artifacts/myapp/versions/1.0.0/extensions/reader [ { "name": "table", "typedescription": "reader", "name": "default",The name of the table to write to. Defaults to 'X'.", "description": "Writes timestamp and body as two columns and expects the row key to come as a header in the stream event.","type": "string", "required": false, "className": "co.cask.cdap.examples.myapp.plugins.DefaultStreamReader" }, "artifact": { "namespacereaderPlugin": "default", { "name": "myappreaderPlugin", "version "description": "1.0.0" } } ] |
It looks like there is only one plugin of type reader available. Another call gives more details about what that plugin requires:
| Code Block |
|---|
GET /namespaces/default/artifacts/myapp/versions/1.0.0/extensions/reader/plugins/default [ { The name of the streamreader plugin to use.", "type": "readerstring", "name": "default", "descriptionrequired": true "Writes timestamp and body as two columns and expects the}, row key to come as a header in the stream event.","readerPluginProperties": { "classNamename": "co.cask.cdap.examples.myapp.plugins.DefaultStreamReader","readerPluginProperties", "propertiesdescription": { "Properties to send to the streamreader plugin.", "rowkeytype": { "plugin", "nameplugintype": "rowkeystreamreader", "descriptionrequired": false "The header that should be used as the row} key to write to. Defaults to 'rowkey'.", } "type": "string",} ], "requiredflows": false[ ... ], } "flowlets": [ ... }], "artifactdatasetModules": [ {... ] } } |
2.2 Adding plugins
A default implementation of the streamreader plugin is created to implement the previous logic:
| Code Block |
|---|
@Plugin(type = "namespacestreamreader":) @Name("default", "name": "myapp", "version": "1.0.0" } } ] |
Now the admin has all the information needed to create an application from the artifact:
| Code Block |
|---|
PUT /namespaces/default/apps/purchaseDump -d ' { "artifact": { "name": "myapp", "version": "1.0.0" }, "config": { "stream": "purchases", "table": "events", "flow": {") @Description("Writes timestamp and body as two columns and expects the row key to come as a header in the stream event.") public class DefaultStreamReader implements StreamReader { private DefaultConfig config; public static class DefaultConfig extends PluginConfig { @Description("The header that should be used as the row key to write to. Defaults to 'rowkey'.") @Nullable private String rowkey; "reader": {private DefaultConfig() { rowkey "name":= "defaultrowkey",; } } "properties": { } public Put read(StreamEvent }, event) { Put put "writer": { ... }= new Put(Bytes.toBytes(event.getHeaders().get(config.rowkey))); }put.add("timestamp", event.getTimestamp()); } }' |
This creates an application that reads from the 'purchases' stream and writes to the 'events' table. In the same way, other applications can be created that read from and write to configurable data sources.
3. Upgrading an Application
A bug is found in the application code. A new version of the artifact is created and deployed:
| Code Block |
|---|
POST /namespaces/default/artifacts/myapp --data-binary @myapp put.add("body", Bytes.toBytes(event.getBody())); return put; } } |
The plugin is bundled into a 'streamreaders-1.0.
...
The admin wants to see what applications are using the old version of the 0.jar' artifact. It is added as an extension to the myapp artifact:
| Code Block |
|---|
GETPOST /namespaces/default/apps?artifactName=myapp&artifactVersion=/artifacts/streamreaders --data-binary streamreaders-1.0.0.jar [ { "name": "purchaseDump" } ] |
The admin stops all running programs for the app using existing APIs. A call is then made to upgrade the app to the latest artifact:
| Code Block |
|---|
PUT-H 'X-Extends-Artifacts: myapp-[2.0.0,3.0.0)' |
The plugin details can now be seen by querying for extensions to myapp:
| Code Block |
|---|
GET /namespaces/default/artifacts/apps/purchaseDump/properties -d ' { "artifact": { "name": "myapp", "version": "1.0.1" }, "config":myapp/versions/2.0.0/extensions [ "streamreader" ] GET /namespaces/default/artifacts/myapp/versions/2.0.0/extensions/streamreader [ { "streamname": "purchasesdefault", "tabletype": "eventsreader", "flowdescription": { "reader":Writes {timestamp and body as two columns and expects the row "name": "default", "properties": { } }, key to come as a header in the stream event.", "writerclassName": { ... }"co.cask.cdap.examples.myapp.plugins.DefaultStreamReader", "properties": }{ } }' |
4. Rolling back an Application
Oops, it turns out v1.0.1 has a critical bug. The admin stops all running programs then makes a call to rollback to the previous version:
| Code Block |
|---|
PUT /namespaces/default/apps/purchaseDump/properties -d ' "rowkey": { "artifact": { "name": "myapprowkey", "versiondescription": "1.0.0" }, "config": { "stream": "purchases",The header that should be used as the row key to write to. Defaults to 'rowkey'.", "tabletype": "eventsstring", "flow": { "readerrequired": {false "name": "default",} } "propertiesartifact": { } }"name": "streamreaders", "writerversion": { "1.0..0" } } } }'] |
5. System Artifacts
System artifacts are special artifacts that can be accessed in other namespaces. They cannot be deployed through the RESTful API unless a conf setting is set. Instead, they are placed in a directory on the CDAP master host. When CDAP starts up, the directory will be scanned and those artifacts will be added to the system. Example uses for system artifacts are the ETLBatch and ETLRealtime applications that we want to include out of the box.
...