diff --git a/CHANGELOG.md b/CHANGELOG.md
index d151afe1e9dc6..30421ebfac4f6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,6 +35,7 @@
   + **Property Changes**
     - AWS::AppSync::DataSource HttpConfig (__added__)
     - AWS::DAX::Cluster SSESpecification (__added__)
+    - AWS::DynamoDB::Table Stream (__added__)
     - AWS::EC2::VPCEndpoint IsPrivateDnsEnabled (__added__)
     - AWS::EC2::VPCEndpoint SecurityGroupIds (__added__)
     - AWS::EC2::VPCEndpoint SubnetIds (__added__)
diff --git a/packages/@aws-cdk/aws-dynamodb/lib/table.ts b/packages/@aws-cdk/aws-dynamodb/lib/table.ts
index d6a3ff419a6f4..d5fdf087c0074 100644
--- a/packages/@aws-cdk/aws-dynamodb/lib/table.ts
+++ b/packages/@aws-cdk/aws-dynamodb/lib/table.ts
@@ -23,6 +23,13 @@ export interface TableProps {
      * @default <generated>
      */
     tableName?: string;
+
+    /**
+     * When an item in the table is modified, StreamViewType determines what information
+     * is written to the stream for this table. Valid values for StreamViewType are:
+     * @default undefined, streams are disbaled
+     */
+    streamSpecification?: StreamViewType;
 }
 
 /**
@@ -48,7 +55,8 @@ export class Table extends Construct {
             tableName: props.tableName,
             keySchema: this.keySchema,
             attributeDefinitions: this.attributeDefinitions,
-            provisionedThroughput: { readCapacityUnits, writeCapacityUnits }
+            provisionedThroughput: { readCapacityUnits, writeCapacityUnits },
+            streamSpecification: props.streamSpecification ? {streamViewType: props.streamSpecification} : undefined
         });
 
         if (props.tableName) { this.addMetadata('aws:cdk:hasPhysicalName', props.tableName); }
@@ -112,3 +120,20 @@ export enum KeyAttributeType {
     Number = 'N',
     String = 'S',
 }
+
+/**
+ * When an item in the table is modified, StreamViewType determines what information
+ * is written to the stream for this table. Valid values for StreamViewType are:
+ * @link https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_StreamSpecification.html
+ * @enum {string}
+ */
+export enum StreamViewType {
+    /** The entire item, as it appears after it was modified, is written to the stream. */
+    NewImage = 'NEW_IMAGE',
+    /** The entire item, as it appeared before it was modified, is written to the stream. */
+    OldImage = 'OLD_IMAGE',
+    /** Both the new and the old item images of the item are written to the stream. */
+    NewAndOldImages = 'NEW_AND_OLD_IMAGES',
+    /** Only the key attributes of the modified item are written to the stream. */
+    KeysOnly = 'KEYS_ONLY'
+  }
diff --git a/packages/@aws-cdk/aws-dynamodb/test/test.dynamodb.ts b/packages/@aws-cdk/aws-dynamodb/test/test.dynamodb.ts
index de6142177eba1..3fda4c4e7bbcf 100644
--- a/packages/@aws-cdk/aws-dynamodb/test/test.dynamodb.ts
+++ b/packages/@aws-cdk/aws-dynamodb/test/test.dynamodb.ts
@@ -1,6 +1,6 @@
 import { App, Stack } from '@aws-cdk/cdk';
 import { Test } from 'nodeunit';
-import { KeyAttributeType, Table } from '../lib';
+import { KeyAttributeType, StreamViewType, Table } from '../lib';
 
 export = {
     'default properties': {
@@ -24,7 +24,7 @@ export = {
                         Properties: {
                             AttributeDefinitions: [{ AttributeName: 'hashKey', AttributeType: 'B' }],
                             KeySchema: [{ AttributeName: 'hashKey', KeyType: 'HASH' }],
-                            ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 }
+                            ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 },
                         }
                     }
                 }
@@ -52,7 +52,7 @@ export = {
                                 { AttributeName: 'hashKey', KeyType: 'HASH' },
                                 { AttributeName: 'sortKey', KeyType: 'RANGE' }
                             ],
-                            ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 }
+                            ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 },
                         }
                     }
                 }
@@ -60,6 +60,139 @@ export = {
 
             test.done();
         },
+        'stream is not enabled by default'(test: Test) {
+            const app = new TestApp();
+            new Table(app.stack, 'MyTable')
+                .addPartitionKey('partitionKey', KeyAttributeType.Binary)
+                .addSortKey('sortKey', KeyAttributeType.Number);
+            const template = app.synthesizeTemplate();
+
+            test.deepEqual(template, {
+                Resources: {
+                    MyTable794EDED1: {
+                        Type: 'AWS::DynamoDB::Table',
+                        Properties: {
+                            AttributeDefinitions: [
+                                { AttributeName: 'partitionKey', AttributeType: 'B' },
+                                { AttributeName: 'sortKey', AttributeType: 'N' }
+                            ],
+                            KeySchema: [
+                                { AttributeName: 'partitionKey', KeyType: 'HASH' },
+                                { AttributeName: 'sortKey', KeyType: 'RANGE' }
+                            ],
+                            ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 },
+                        }
+                    }
+                }
+            });
+
+            test.done();
+        },
+        'can specify new and old images'(test: Test) {
+            const app = new TestApp();
+            const table = new Table(app.stack, 'MyTable', {
+                tableName: 'MyTable',
+                readCapacity: 42,
+                writeCapacity: 1337,
+                streamSpecification: StreamViewType.NewAndOldImages
+            });
+            table.addPartitionKey('partitionKey', KeyAttributeType.String);
+            table.addSortKey('sortKey', KeyAttributeType.Binary);
+            const template = app.synthesizeTemplate();
+
+            test.deepEqual(template, {
+                Resources: {
+                    MyTable794EDED1: {
+                        Type: 'AWS::DynamoDB::Table',
+                        Properties: {
+                            AttributeDefinitions: [
+                                { AttributeName: 'partitionKey', AttributeType: 'S' },
+                                { AttributeName: 'sortKey', AttributeType: 'B' }
+                            ],
+                            StreamSpecification: { StreamViewType: 'NEW_AND_OLD_IMAGES' },
+                            KeySchema: [
+                                { AttributeName: 'partitionKey', KeyType: 'HASH' },
+                                { AttributeName: 'sortKey', KeyType: 'RANGE' }
+                            ],
+                            ProvisionedThroughput: { ReadCapacityUnits: 42, WriteCapacityUnits: 1337 },
+                            TableName: 'MyTable'
+                        }
+                    }
+                }
+            });
+
+            test.done();
+        },
+        'can specify new images only'(test: Test) {
+            const app = new TestApp();
+            const table = new Table(app.stack, 'MyTable', {
+                tableName: 'MyTable',
+                readCapacity: 42,
+                writeCapacity: 1337,
+                streamSpecification: StreamViewType.NewImage
+            });
+            table.addPartitionKey('partitionKey', KeyAttributeType.String);
+            table.addSortKey('sortKey', KeyAttributeType.Binary);
+            const template = app.synthesizeTemplate();
+
+            test.deepEqual(template, {
+                Resources: {
+                    MyTable794EDED1: {
+                        Type: 'AWS::DynamoDB::Table',
+                        Properties: {
+                            KeySchema: [
+                                { AttributeName: 'partitionKey', KeyType: 'HASH' },
+                                { AttributeName: 'sortKey', KeyType: 'RANGE' }
+                            ],
+                            ProvisionedThroughput: { ReadCapacityUnits: 42, WriteCapacityUnits: 1337 },
+                            AttributeDefinitions: [
+                                { AttributeName: 'partitionKey', AttributeType: 'S' },
+                                { AttributeName: 'sortKey', AttributeType: 'B' }
+                            ],
+                            StreamSpecification: { StreamViewType: 'NEW_IMAGE' },
+                            TableName: 'MyTable'
+                        }
+                    }
+                }
+            });
+
+            test.done();
+        },
+        'can specify old images only'(test: Test) {
+            const app = new TestApp();
+            const table = new Table(app.stack, 'MyTable', {
+                tableName: 'MyTable',
+                readCapacity: 42,
+                writeCapacity: 1337,
+                streamSpecification: StreamViewType.OldImage
+            });
+            table.addPartitionKey('partitionKey', KeyAttributeType.String);
+            table.addSortKey('sortKey', KeyAttributeType.Binary);
+            const template = app.synthesizeTemplate();
+
+            test.deepEqual(template, {
+                Resources: {
+                    MyTable794EDED1: {
+                        Type: 'AWS::DynamoDB::Table',
+                        Properties: {
+                            KeySchema: [
+                                { AttributeName: 'partitionKey', KeyType: 'HASH' },
+                                { AttributeName: 'sortKey', KeyType: 'RANGE' }
+                            ],
+                            ProvisionedThroughput: { ReadCapacityUnits: 42, WriteCapacityUnits: 1337 },
+                            AttributeDefinitions: [
+                                { AttributeName: 'partitionKey', AttributeType: 'S' },
+                                { AttributeName: 'sortKey', AttributeType: 'B' }
+                            ],
+                            StreamSpecification: { StreamViewType: 'OLD_IMAGE' },
+                            TableName: 'MyTable'
+                        }
+                    }
+                }
+            });
+
+            test.done();
+        }
     },
 
     'when specifying every property'(test: Test) {
@@ -90,7 +223,7 @@ export = {
                             ReadCapacityUnits: 42,
                             WriteCapacityUnits: 1337
                         },
-                        TableName: 'MyTable'
+                        TableName: 'MyTable',
                     }
                 }
             }