diff --git a/packages/@aws-cdk/aws-rds/README.md b/packages/@aws-cdk/aws-rds/README.md
index 1d0bfbe9b947a..fb0890c336402 100644
--- a/packages/@aws-cdk/aws-rds/README.md
+++ b/packages/@aws-cdk/aws-rds/README.md
@@ -62,6 +62,18 @@ By default, the master password will be generated and stored in AWS Secrets Mana
 Your cluster will be empty by default. To add a default database upon construction, specify the
 `defaultDatabaseName` attribute.
 
+Use `DatabaseClusterFromSnapshot` to create a cluster from a snapshot:
+
+```ts
+new DatabaseClusterFromSnapshot(stack, 'Database', {
+  engine: DatabaseClusterEngine.aurora({ version: AuroraEngineVersion.VER_1_22_2 }),
+  instanceProps: {
+    vpc,
+  },
+  snapshotIdentifier: 'mySnapshot',
+});
+```
+
 ### Starting an instance database
 
 To set up a instance database, define a `DatabaseInstance`. You must
diff --git a/packages/@aws-cdk/aws-rds/lib/cluster.ts b/packages/@aws-cdk/aws-rds/lib/cluster.ts
index 4ddba423ba8ea..89b98756e0b8b 100644
--- a/packages/@aws-cdk/aws-rds/lib/cluster.ts
+++ b/packages/@aws-cdk/aws-rds/lib/cluster.ts
@@ -12,12 +12,12 @@ import { Endpoint } from './endpoint';
 import { IParameterGroup } from './parameter-group';
 import { BackupProps, InstanceProps, Login, PerformanceInsightRetention, RotationMultiUserOptions } from './props';
 import { DatabaseProxy, DatabaseProxyOptions, ProxyTarget } from './proxy';
-import { CfnDBCluster, CfnDBInstance, CfnDBSubnetGroup } from './rds.generated';
+import { CfnDBCluster, CfnDBClusterProps, CfnDBInstance, CfnDBSubnetGroup } from './rds.generated';
 
 /**
- * Properties for a new database cluster
+ * Common properties for a new database cluster or cluster from snapshot.
  */
-export interface DatabaseClusterProps {
+interface DatabaseClusterBaseProps {
   /**
    * What kind of database to start
    */
@@ -37,11 +37,6 @@ export interface DatabaseClusterProps {
    */
   readonly instanceProps: InstanceProps;
 
-  /**
-   * Username and password for the administrative user
-   */
-  readonly masterUser: Login;
-
   /**
    * Backup settings
    *
@@ -90,21 +85,6 @@ export interface DatabaseClusterProps {
    */
   readonly deletionProtection?: boolean;
 
-  /**
-   * Whether to enable storage encryption.
-   *
-   * @default - true if storageEncryptionKey is provided, false otherwise
-   */
-  readonly storageEncrypted?: boolean
-
-  /**
-   * The KMS key for storage encryption.
-   * If specified, {@link storageEncrypted} will be set to `true`.
-   *
-   * @default - if storageEncrypted is true then the default master key, no key otherwise
-   */
-  readonly storageEncryptionKey?: kms.IKey;
-
   /**
    * A preferred maintenance window day/time range. Should be specified as a range ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC).
    *
@@ -289,87 +269,20 @@ abstract class DatabaseClusterBase extends Resource implements IDatabaseCluster
 }
 
 /**
- * Create a clustered database with a given number of instances.
- *
- * @resource AWS::RDS::DBCluster
+ * Abstract base for ``DatabaseCluster`` and ``DatabaseClusterFromSnapshot``
  */
-export class DatabaseCluster extends DatabaseClusterBase {
-  /**
-   * Import an existing DatabaseCluster from properties
-   */
-  public static fromDatabaseClusterAttributes(scope: Construct, id: string, attrs: DatabaseClusterAttributes): IDatabaseCluster {
-    class Import extends DatabaseClusterBase implements IDatabaseCluster {
-      public readonly defaultPort = ec2.Port.tcp(attrs.port);
-      public readonly connections = new ec2.Connections({
-        securityGroups: attrs.securityGroups,
-        defaultPort: this.defaultPort,
-      });
-      public readonly clusterIdentifier = attrs.clusterIdentifier;
-      public readonly instanceIdentifiers: string[] = [];
-      public readonly clusterEndpoint = new Endpoint(attrs.clusterEndpointAddress, attrs.port);
-      public readonly clusterReadEndpoint = new Endpoint(attrs.readerEndpointAddress, attrs.port);
-      public readonly instanceEndpoints = attrs.instanceEndpointAddresses.map(a => new Endpoint(a, attrs.port));
-    }
-
-    return new Import(scope, id);
-  }
+abstract class DatabaseClusterNew extends DatabaseClusterBase {
 
-  /**
-   * Identifier of the cluster
-   */
-  public readonly clusterIdentifier: string;
-
-  /**
-   * Identifiers of the replicas
-   */
   public readonly instanceIdentifiers: string[] = [];
-
-  /**
-   * The endpoint to use for read/write operations
-   */
-  public readonly clusterEndpoint: Endpoint;
-
-  /**
-   * Endpoint to use for load-balanced read-only operations.
-   */
-  public readonly clusterReadEndpoint: Endpoint;
-
-  /**
-   * Endpoints which address each individual replica.
-   */
   public readonly instanceEndpoints: Endpoint[] = [];
 
-  /**
-   * Access to the network connections
-   */
-  public readonly connections: ec2.Connections;
-
-  /**
-   * The secret attached to this cluster
-   */
-  public readonly secret?: secretsmanager.ISecret;
-
-  private readonly singleUserRotationApplication: secretsmanager.SecretRotationApplication;
-  private readonly multiUserRotationApplication: secretsmanager.SecretRotationApplication;
-
-  /**
-   * The VPC where the DB subnet group is created.
-   */
-  private readonly vpc: ec2.IVpc;
+  protected readonly newCfnProps: CfnDBClusterProps;
+  protected readonly subnetGroup: CfnDBSubnetGroup;
+  protected readonly securityGroups: ec2.ISecurityGroup[];
 
-  /**
-   * The subnets used by the DB subnet group.
-   *
-   * @default - the Vpc default strategy if not specified.
-   */
-  private readonly vpcSubnets?: ec2.SubnetSelection;
-
-  constructor(scope: Construct, id: string, props: DatabaseClusterProps) {
+  constructor(scope: Construct, id: string, props: DatabaseClusterBaseProps) {
     super(scope, id);
 
-    this.vpc = props.instanceProps.vpc;
-    this.vpcSubnets = props.instanceProps.vpcSubnets;
-
     const { subnetIds } = props.instanceProps.vpc.selectSubnets(props.instanceProps.vpcSubnets);
 
     // Cannot test whether the subnets are in different AZs, but at least we can test the amount.
@@ -377,32 +290,21 @@ export class DatabaseCluster extends DatabaseClusterBase {
       this.node.addError(`Cluster requires at least 2 subnets, got ${subnetIds.length}`);
     }
 
-    const subnetGroup = new CfnDBSubnetGroup(this, 'Subnets', {
+    this.subnetGroup = new CfnDBSubnetGroup(this, 'Subnets', {
       dbSubnetGroupDescription: `Subnets for ${id} database`,
       subnetIds,
     });
     if (props.removalPolicy === RemovalPolicy.RETAIN) {
-      subnetGroup.applyRemovalPolicy(RemovalPolicy.RETAIN);
+      this.subnetGroup.applyRemovalPolicy(RemovalPolicy.RETAIN);
     }
 
-    const securityGroups = props.instanceProps.securityGroups ?? [
+    this.securityGroups = props.instanceProps.securityGroups ?? [
       new ec2.SecurityGroup(this, 'SecurityGroup', {
         description: 'RDS security group',
         vpc: props.instanceProps.vpc,
       }),
     ];
 
-    let secret: DatabaseSecret | undefined;
-    if (!props.masterUser.password) {
-      secret = new DatabaseSecret(this, 'Secret', {
-        username: props.masterUser.username,
-        encryptionKey: props.masterUser.encryptionKey,
-      });
-    }
-
-    this.singleUserRotationApplication = props.engine.singleUserRotationApplication;
-    this.multiUserRotationApplication = props.engine.multiUserRotationApplication;
-
     const clusterAssociatedRoles: CfnDBCluster.DBClusterRoleProperty[] = [];
     let { s3ImportRole, s3ExportRole } = this.setupS3ImportExport(props);
     if (s3ImportRole) {
@@ -421,44 +323,169 @@ export class DatabaseCluster extends DatabaseClusterBase {
     const clusterParameterGroup = props.parameterGroup ?? clusterEngineBindConfig.parameterGroup;
     const clusterParameterGroupConfig = clusterParameterGroup?.bindToCluster({});
 
-    const cluster = new CfnDBCluster(this, 'Resource', {
+    this.newCfnProps = {
       // Basic
       engine: props.engine.engineType,
       engineVersion: props.engine.engineVersion?.fullVersion,
       dbClusterIdentifier: props.clusterIdentifier,
-      dbSubnetGroupName: subnetGroup.ref,
-      vpcSecurityGroupIds: securityGroups.map(sg => sg.securityGroupId),
+      dbSubnetGroupName: this.subnetGroup.ref,
+      vpcSecurityGroupIds: this.securityGroups.map(sg => sg.securityGroupId),
       port: props.port ?? clusterEngineBindConfig.port,
       dbClusterParameterGroupName: clusterParameterGroupConfig?.parameterGroupName,
       associatedRoles: clusterAssociatedRoles.length > 0 ? clusterAssociatedRoles : undefined,
       deletionProtection: props.deletionProtection,
       // Admin
-      masterUsername: secret ? secret.secretValueFromJson('username').toString() : props.masterUser.username,
-      masterUserPassword: secret
-        ? secret.secretValueFromJson('password').toString()
-        : (props.masterUser.password
-          ? props.masterUser.password.toString()
-          : undefined),
       backupRetentionPeriod: props.backup?.retention?.toDays(),
       preferredBackupWindow: props.backup?.preferredWindow,
       preferredMaintenanceWindow: props.preferredMaintenanceWindow,
       databaseName: props.defaultDatabaseName,
       enableCloudwatchLogsExports: props.cloudwatchLogsExports,
-      // Encryption
-      kmsKeyId: props.storageEncryptionKey?.keyArn,
-      storageEncrypted: props.storageEncryptionKey ? true : props.storageEncrypted,
-    });
+    };
+  }
 
+  protected setRemovalPolicy(cluster: CfnDBCluster, removalPolicy?: RemovalPolicy) {
     // if removalPolicy was not specified,
     // leave it as the default, which is Snapshot
-    if (props.removalPolicy) {
-      cluster.applyRemovalPolicy(props.removalPolicy);
+    if (removalPolicy) {
+      cluster.applyRemovalPolicy(removalPolicy);
     } else {
       // The CFN default makes sense for DeletionPolicy,
       // but doesn't cover UpdateReplacePolicy.
       // Fix that here.
       cluster.cfnOptions.updateReplacePolicy = CfnDeletionPolicy.SNAPSHOT;
     }
+  }
+
+  private setupS3ImportExport(props: DatabaseClusterBaseProps): { s3ImportRole?: IRole, s3ExportRole?: IRole } {
+    let s3ImportRole = props.s3ImportRole;
+    if (props.s3ImportBuckets && props.s3ImportBuckets.length > 0) {
+      if (props.s3ImportRole) {
+        throw new Error('Only one of s3ImportRole or s3ImportBuckets must be specified, not both.');
+      }
+
+      s3ImportRole = new Role(this, 'S3ImportRole', {
+        assumedBy: new ServicePrincipal('rds.amazonaws.com'),
+      });
+      for (const bucket of props.s3ImportBuckets) {
+        bucket.grantRead(s3ImportRole);
+      }
+    }
+
+    let s3ExportRole = props.s3ExportRole;
+    if (props.s3ExportBuckets && props.s3ExportBuckets.length > 0) {
+      if (props.s3ExportRole) {
+        throw new Error('Only one of s3ExportRole or s3ExportBuckets must be specified, not both.');
+      }
+
+      s3ExportRole = new Role(this, 'S3ExportRole', {
+        assumedBy: new ServicePrincipal('rds.amazonaws.com'),
+      });
+      for (const bucket of props.s3ExportBuckets) {
+        bucket.grantReadWrite(s3ExportRole);
+      }
+    }
+
+    return { s3ImportRole, s3ExportRole };
+  }
+}
+
+/**
+ * Properties for a new database cluster
+ */
+export interface DatabaseClusterProps extends DatabaseClusterBaseProps {
+  /**
+   * Username and password for the administrative user
+   */
+  readonly masterUser: Login;
+
+  /**
+   * Whether to enable storage encryption.
+   *
+   * @default - true if storageEncryptionKey is provided, false otherwise
+   */
+  readonly storageEncrypted?: boolean
+
+  /**
+   * The KMS key for storage encryption.
+   * If specified, {@link storageEncrypted} will be set to `true`.
+   *
+   * @default - if storageEncrypted is true then the default master key, no key otherwise
+   */
+  readonly storageEncryptionKey?: kms.IKey;
+}
+
+/**
+ * Create a clustered database with a given number of instances.
+ *
+ * @resource AWS::RDS::DBCluster
+ */
+export class DatabaseCluster extends DatabaseClusterNew {
+  /**
+   * Import an existing DatabaseCluster from properties
+   */
+  public static fromDatabaseClusterAttributes(scope: Construct, id: string, attrs: DatabaseClusterAttributes): IDatabaseCluster {
+    class Import extends DatabaseClusterBase implements IDatabaseCluster {
+      public readonly defaultPort = ec2.Port.tcp(attrs.port);
+      public readonly connections = new ec2.Connections({
+        securityGroups: attrs.securityGroups,
+        defaultPort: this.defaultPort,
+      });
+      public readonly clusterIdentifier = attrs.clusterIdentifier;
+      public readonly instanceIdentifiers: string[] = [];
+      public readonly clusterEndpoint = new Endpoint(attrs.clusterEndpointAddress, attrs.port);
+      public readonly clusterReadEndpoint = new Endpoint(attrs.readerEndpointAddress, attrs.port);
+      public readonly instanceEndpoints = attrs.instanceEndpointAddresses.map(a => new Endpoint(a, attrs.port));
+    }
+
+    return new Import(scope, id);
+  }
+
+  public readonly clusterIdentifier: string;
+  public readonly clusterEndpoint: Endpoint;
+  public readonly clusterReadEndpoint: Endpoint;
+  public readonly connections: ec2.Connections;
+
+  /**
+   * The secret attached to this cluster
+   */
+  public readonly secret?: secretsmanager.ISecret;
+
+  private readonly vpc: ec2.IVpc;
+  private readonly vpcSubnets?: ec2.SubnetSelection;
+
+  private readonly singleUserRotationApplication: secretsmanager.SecretRotationApplication;
+  private readonly multiUserRotationApplication: secretsmanager.SecretRotationApplication;
+
+  constructor(scope: Construct, id: string, props: DatabaseClusterProps) {
+    super(scope, id, props);
+
+    this.vpc = props.instanceProps.vpc;
+    this.vpcSubnets = props.instanceProps.vpcSubnets;
+
+    this.singleUserRotationApplication = props.engine.singleUserRotationApplication;
+    this.multiUserRotationApplication = props.engine.multiUserRotationApplication;
+
+    let secret: DatabaseSecret | undefined;
+    if (!props.masterUser.password) {
+      secret = new DatabaseSecret(this, 'Secret', {
+        username: props.masterUser.username,
+        encryptionKey: props.masterUser.encryptionKey,
+      });
+    }
+
+    const cluster = new CfnDBCluster(this, 'Resource', {
+      ...this.newCfnProps,
+      // Admin
+      masterUsername: secret ? secret.secretValueFromJson('username').toString() : props.masterUser.username,
+      masterUserPassword: secret
+        ? secret.secretValueFromJson('password').toString()
+        : (props.masterUser.password
+          ? props.masterUser.password.toString()
+          : undefined),
+      // Encryption
+      kmsKeyId: props.storageEncryptionKey?.keyArn,
+      storageEncrypted: props.storageEncryptionKey ? true : props.storageEncrypted,
+    });
 
     this.clusterIdentifier = cluster.ref;
 
@@ -466,20 +493,21 @@ export class DatabaseCluster extends DatabaseClusterBase {
     const portAttribute = Token.asNumber(cluster.attrEndpointPort);
     this.clusterEndpoint = new Endpoint(cluster.attrEndpointAddress, portAttribute);
     this.clusterReadEndpoint = new Endpoint(cluster.attrReadEndpointAddress, portAttribute);
+    this.connections = new ec2.Connections({
+      securityGroups: this.securityGroups,
+      defaultPort: ec2.Port.tcp(this.clusterEndpoint.port),
+    });
 
-    this.setLogRetention(props);
+    this.setRemovalPolicy(cluster, props.removalPolicy);
 
     if (secret) {
       this.secret = secret.attach(this);
     }
 
-    this.createInstances(props, cluster, subnetGroup, portAttribute);
-
-    const defaultPort = ec2.Port.tcp(this.clusterEndpoint.port);
-    this.connections = new ec2.Connections({ securityGroups, defaultPort });
+    setLogRetention(this, props);
+    createInstances(this, props, this.subnetGroup);
   }
 
-
   /**
    * Adds the single user rotation of the master password to this cluster.
    *
@@ -524,131 +552,169 @@ export class DatabaseCluster extends DatabaseClusterBase {
       target: this,
     });
   }
+}
 
-  private setupS3ImportExport(props: DatabaseClusterProps): { s3ImportRole?: IRole, s3ExportRole?: IRole } {
-    let s3ImportRole = props.s3ImportRole;
-    if (props.s3ImportBuckets && props.s3ImportBuckets.length > 0) {
-      if (props.s3ImportRole) {
-        throw new Error('Only one of s3ImportRole or s3ImportBuckets must be specified, not both.');
-      }
+/**
+ * Properties for ``DatabaseClusterFromSnapshot``
+ */
+export interface DatabaseClusterFromSnapshotProps extends DatabaseClusterBaseProps {
+  /**
+   * The identifier for the DB instance snapshot or DB cluster snapshot to restore from.
+   * You can use either the name or the Amazon Resource Name (ARN) to specify a DB cluster snapshot.
+   * However, you can use only the ARN to specify a DB instance snapshot.
+   */
+  readonly snapshotIdentifier: string;
+}
 
-      s3ImportRole = new Role(this, 'S3ImportRole', {
-        assumedBy: new ServicePrincipal('rds.amazonaws.com'),
-      });
-      for (const bucket of props.s3ImportBuckets) {
-        bucket.grantRead(s3ImportRole);
-      }
-    }
+/**
+ * A database cluster restored from a snapshot.
+ *
+ * @resource AWS::RDS::DBInstance
+ */
+export class DatabaseClusterFromSnapshot extends DatabaseClusterNew {
+  public readonly clusterIdentifier: string;
+  public readonly clusterEndpoint: Endpoint;
+  public readonly clusterReadEndpoint: Endpoint;
+  public readonly connections: ec2.Connections;
 
-    let s3ExportRole = props.s3ExportRole;
-    if (props.s3ExportBuckets && props.s3ExportBuckets.length > 0) {
-      if (props.s3ExportRole) {
-        throw new Error('Only one of s3ExportRole or s3ExportBuckets must be specified, not both.');
-      }
+  constructor(scope: Construct, id: string, props: DatabaseClusterFromSnapshotProps) {
+    super(scope, id, props);
 
-      s3ExportRole = new Role(this, 'S3ExportRole', {
-        assumedBy: new ServicePrincipal('rds.amazonaws.com'),
-      });
-      for (const bucket of props.s3ExportBuckets) {
-        bucket.grantReadWrite(s3ExportRole);
-      }
-    }
+    const cluster = new CfnDBCluster(this, 'Resource', {
+      ...this.newCfnProps,
+      snapshotIdentifier: props.snapshotIdentifier,
+    });
 
-    return { s3ImportRole, s3ExportRole };
-  }
+    this.clusterIdentifier = cluster.ref;
 
-  private createInstances(props: DatabaseClusterProps, cluster: CfnDBCluster, subnetGroup: CfnDBSubnetGroup, portAttribute: number) {
-    const instanceCount = props.instances != null ? props.instances : 2;
-    if (instanceCount < 1) {
-      throw new Error('At least one instance is required');
-    }
+    // create a number token that represents the port of the cluster
+    const portAttribute = Token.asNumber(cluster.attrEndpointPort);
+    this.clusterEndpoint = new Endpoint(cluster.attrEndpointAddress, portAttribute);
+    this.clusterReadEndpoint = new Endpoint(cluster.attrReadEndpointAddress, portAttribute);
+    this.connections = new ec2.Connections({
+      securityGroups: this.securityGroups,
+      defaultPort: ec2.Port.tcp(this.clusterEndpoint.port),
+    });
 
-    const instanceProps = props.instanceProps;
-    // Get the actual subnet objects so we can depend on internet connectivity.
-    const internetConnected = instanceProps.vpc.selectSubnets(instanceProps.vpcSubnets).internetConnectivityEstablished;
-
-    let monitoringRole;
-    if (props.monitoringInterval && props.monitoringInterval.toSeconds()) {
-      monitoringRole = props.monitoringRole || new Role(this, 'MonitoringRole', {
-        assumedBy: new ServicePrincipal('monitoring.rds.amazonaws.com'),
-        managedPolicies: [
-          ManagedPolicy.fromAwsManagedPolicyName('service-role/AmazonRDSEnhancedMonitoringRole'),
-        ],
-      });
+    this.setRemovalPolicy(cluster, props.removalPolicy);
+
+    setLogRetention(this, props);
+    createInstances(this, props, this.subnetGroup);
+  }
+}
+
+/**
+ * Sets up CloudWatch log retention if configured.
+ * A function rather than protected member to prevent exposing ``DatabaseClusterBaseProps``.
+ */
+function setLogRetention(cluster: DatabaseClusterNew, props: DatabaseClusterBaseProps) {
+  if (props.cloudwatchLogsExports) {
+    const unsupportedLogTypes = props.cloudwatchLogsExports.filter(logType => !props.engine.supportedLogTypes.includes(logType));
+    if (unsupportedLogTypes.length > 0) {
+      throw new Error(`Unsupported logs for the current engine type: ${unsupportedLogTypes.join(',')}`);
     }
 
-    const enablePerformanceInsights = instanceProps.enablePerformanceInsights
-      || instanceProps.performanceInsightRetention !== undefined || instanceProps.performanceInsightEncryptionKey !== undefined;
-    if (enablePerformanceInsights && instanceProps.enablePerformanceInsights === false) {
-      throw new Error('`enablePerformanceInsights` disabled, but `performanceInsightRetention` or `performanceInsightEncryptionKey` was set');
+    if (props.cloudwatchLogsRetention) {
+      for (const log of props.cloudwatchLogsExports) {
+        new logs.LogRetention(cluster, `LogRetention${log}`, {
+          logGroupName: `/aws/rds/cluster/${cluster.clusterIdentifier}/${log}`,
+          retention: props.cloudwatchLogsRetention,
+          role: props.cloudwatchLogsRetentionRole,
+        });
+      }
     }
+  }
+}
 
-    const instanceType = instanceProps.instanceType ?? ec2.InstanceType.of(ec2.InstanceClass.T3, ec2.InstanceSize.MEDIUM);
-    const instanceParameterGroupConfig = instanceProps.parameterGroup?.bindToInstance({});
-    for (let i = 0; i < instanceCount; i++) {
-      const instanceIndex = i + 1;
-      const instanceIdentifier = props.instanceIdentifierBase != null ? `${props.instanceIdentifierBase}${instanceIndex}` :
-        props.clusterIdentifier != null ? `${props.clusterIdentifier}instance${instanceIndex}` :
-          undefined;
-
-      const publiclyAccessible = instanceProps.vpcSubnets && instanceProps.vpcSubnets.subnetType === ec2.SubnetType.PUBLIC;
-
-      const instance = new CfnDBInstance(this, `Instance${instanceIndex}`, {
-        // Link to cluster
-        engine: props.engine.engineType,
-        engineVersion: props.engine.engineVersion?.fullVersion,
-        dbClusterIdentifier: cluster.ref,
-        dbInstanceIdentifier: instanceIdentifier,
-        // Instance properties
-        dbInstanceClass: databaseInstanceType(instanceType),
-        publiclyAccessible,
-        enablePerformanceInsights: enablePerformanceInsights || instanceProps.enablePerformanceInsights, // fall back to undefined if not set
-        performanceInsightsKmsKeyId: instanceProps.performanceInsightEncryptionKey?.keyArn,
-        performanceInsightsRetentionPeriod: enablePerformanceInsights
-          ? (instanceProps.performanceInsightRetention || PerformanceInsightRetention.DEFAULT)
-          : undefined,
-        // This is already set on the Cluster. Unclear to me whether it should be repeated or not. Better yes.
-        dbSubnetGroupName: subnetGroup.ref,
-        dbParameterGroupName: instanceParameterGroupConfig?.parameterGroupName,
-        monitoringInterval: props.monitoringInterval && props.monitoringInterval.toSeconds(),
-        monitoringRoleArn: monitoringRole && monitoringRole.roleArn,
-      });
+/** Output from the createInstances method; used to set instance identifiers and endpoints */
+interface InstanceConfig {
+  readonly instanceIdentifiers: string[];
+  readonly instanceEndpoints: Endpoint[];
+}
 
-      // If removalPolicy isn't explicitly set,
-      // it's Snapshot for Cluster.
-      // Because of that, in this case,
-      // we can safely use the CFN default of Delete for DbInstances with dbClusterIdentifier set.
-      if (props.removalPolicy) {
-        instance.applyRemovalPolicy(props.removalPolicy);
-      }
+/**
+ * Creates the instances for the cluster.
+ * A function rather than a protected method on ``DatabaseClusterNew`` to avoid exposing
+ * ``DatabaseClusterNew`` and ``DatabaseClusterBaseProps`` in the API.
+ */
+function createInstances(cluster: DatabaseClusterNew, props: DatabaseClusterBaseProps, subnetGroup: CfnDBSubnetGroup): InstanceConfig {
+  const instanceCount = props.instances != null ? props.instances : 2;
+  if (instanceCount < 1) {
+    throw new Error('At least one instance is required');
+  }
 
-      // We must have a dependency on the NAT gateway provider here to create
-      // things in the right order.
-      instance.node.addDependency(internetConnected);
+  const instanceIdentifiers: string[] = [];
+  const instanceEndpoints: Endpoint[] = [];
+  const portAttribute = cluster.clusterEndpoint.port;
+  const instanceProps = props.instanceProps;
+
+  // Get the actual subnet objects so we can depend on internet connectivity.
+  const internetConnected = instanceProps.vpc.selectSubnets(instanceProps.vpcSubnets).internetConnectivityEstablished;
+
+  let monitoringRole;
+  if (props.monitoringInterval && props.monitoringInterval.toSeconds()) {
+    monitoringRole = props.monitoringRole || new Role(cluster, 'MonitoringRole', {
+      assumedBy: new ServicePrincipal('monitoring.rds.amazonaws.com'),
+      managedPolicies: [
+        ManagedPolicy.fromAwsManagedPolicyName('service-role/AmazonRDSEnhancedMonitoringRole'),
+      ],
+    });
+  }
 
-      this.instanceIdentifiers.push(instance.ref);
-      this.instanceEndpoints.push(new Endpoint(instance.attrEndpointAddress, portAttribute));
-    }
+  const enablePerformanceInsights = instanceProps.enablePerformanceInsights
+    || instanceProps.performanceInsightRetention !== undefined || instanceProps.performanceInsightEncryptionKey !== undefined;
+  if (enablePerformanceInsights && instanceProps.enablePerformanceInsights === false) {
+    throw new Error('`enablePerformanceInsights` disabled, but `performanceInsightRetention` or `performanceInsightEncryptionKey` was set');
   }
 
-  private setLogRetention(props: DatabaseClusterProps) {
-    if (props.cloudwatchLogsExports) {
-      const unsupportedLogTypes = props.cloudwatchLogsExports.filter(logType => !props.engine.supportedLogTypes.includes(logType));
-      if (unsupportedLogTypes.length > 0) {
-        throw new Error(`Unsupported logs for the current engine type: ${unsupportedLogTypes.join(',')}`);
-      }
+  const instanceType = instanceProps.instanceType ?? ec2.InstanceType.of(ec2.InstanceClass.T3, ec2.InstanceSize.MEDIUM);
+  const instanceParameterGroupConfig = instanceProps.parameterGroup?.bindToInstance({});
+  for (let i = 0; i < instanceCount; i++) {
+    const instanceIndex = i + 1;
+    const instanceIdentifier = props.instanceIdentifierBase != null ? `${props.instanceIdentifierBase}${instanceIndex}` :
+      props.clusterIdentifier != null ? `${props.clusterIdentifier}instance${instanceIndex}` :
+        undefined;
 
-      if (props.cloudwatchLogsRetention) {
-        for (const log of props.cloudwatchLogsExports) {
-          new logs.LogRetention(this, `LogRetention${log}`, {
-            logGroupName: `/aws/rds/cluster/${this.clusterIdentifier}/${log}`,
-            retention: props.cloudwatchLogsRetention,
-            role: props.cloudwatchLogsRetentionRole,
-          });
-        }
-      }
+    const publiclyAccessible = instanceProps.vpcSubnets && instanceProps.vpcSubnets.subnetType === ec2.SubnetType.PUBLIC;
+
+    const instance = new CfnDBInstance(cluster, `Instance${instanceIndex}`, {
+      // Link to cluster
+      engine: props.engine.engineType,
+      engineVersion: props.engine.engineVersion?.fullVersion,
+      dbClusterIdentifier: cluster.clusterIdentifier,
+      dbInstanceIdentifier: instanceIdentifier,
+      // Instance properties
+      dbInstanceClass: databaseInstanceType(instanceType),
+      publiclyAccessible,
+      enablePerformanceInsights: enablePerformanceInsights || instanceProps.enablePerformanceInsights, // fall back to undefined if not set
+      performanceInsightsKmsKeyId: instanceProps.performanceInsightEncryptionKey?.keyArn,
+      performanceInsightsRetentionPeriod: enablePerformanceInsights
+        ? (instanceProps.performanceInsightRetention || PerformanceInsightRetention.DEFAULT)
+        : undefined,
+      // This is already set on the Cluster. Unclear to me whether it should be repeated or not. Better yes.
+      dbSubnetGroupName: subnetGroup.ref,
+      dbParameterGroupName: instanceParameterGroupConfig?.parameterGroupName,
+      monitoringInterval: props.monitoringInterval && props.monitoringInterval.toSeconds(),
+      monitoringRoleArn: monitoringRole && monitoringRole.roleArn,
+    });
+
+    // If removalPolicy isn't explicitly set,
+    // it's Snapshot for Cluster.
+    // Because of that, in this case,
+    // we can safely use the CFN default of Delete for DbInstances with dbClusterIdentifier set.
+    if (props.removalPolicy) {
+      instance.applyRemovalPolicy(props.removalPolicy);
     }
+
+    // We must have a dependency on the NAT gateway provider here to create
+    // things in the right order.
+    instance.node.addDependency(internetConnected);
+
+    instanceIdentifiers.push(instance.ref);
+    instanceEndpoints.push(new Endpoint(instance.attrEndpointAddress, portAttribute));
   }
+
+  return { instanceEndpoints, instanceIdentifiers };
 }
 
 /**
diff --git a/packages/@aws-cdk/aws-rds/package.json b/packages/@aws-cdk/aws-rds/package.json
index 7c65dfad99e40..90a92079eb6e5 100644
--- a/packages/@aws-cdk/aws-rds/package.json
+++ b/packages/@aws-cdk/aws-rds/package.json
@@ -105,6 +105,7 @@
     "exclude": [
       "props-physical-name:@aws-cdk/aws-rds.ParameterGroupProps",
       "props-physical-name:@aws-cdk/aws-rds.DatabaseClusterProps",
+      "props-physical-name:@aws-cdk/aws-rds.DatabaseClusterFromSnapshotProps",
       "props-physical-name:@aws-cdk/aws-rds.DatabaseInstanceProps",
       "props-physical-name:@aws-cdk/aws-rds.DatabaseInstanceFromSnapshotProps",
       "props-physical-name:@aws-cdk/aws-rds.DatabaseInstanceReadReplicaProps",
diff --git a/packages/@aws-cdk/aws-rds/test/test.cluster.ts b/packages/@aws-cdk/aws-rds/test/test.cluster.ts
index 09637a8deea1c..fa8480e000f5d 100644
--- a/packages/@aws-cdk/aws-rds/test/test.cluster.ts
+++ b/packages/@aws-cdk/aws-rds/test/test.cluster.ts
@@ -6,7 +6,10 @@ import * as logs from '@aws-cdk/aws-logs';
 import * as s3 from '@aws-cdk/aws-s3';
 import * as cdk from '@aws-cdk/core';
 import { Test } from 'nodeunit';
-import { AuroraMysqlEngineVersion, AuroraPostgresEngineVersion, DatabaseCluster, DatabaseClusterEngine, ParameterGroup, PerformanceInsightRetention } from '../lib';
+import {
+  AuroraEngineVersion, AuroraMysqlEngineVersion, AuroraPostgresEngineVersion, DatabaseCluster, DatabaseClusterEngine,
+  DatabaseClusterFromSnapshot, ParameterGroup, PerformanceInsightRetention,
+} from '../lib';
 
 export = {
   'creating a Cluster also creates 2 DB Instances'(test: Test) {
@@ -1310,6 +1313,37 @@ export = {
 
     test.done();
   },
+
+  'create a cluster from a snapshot'(test: Test) {
+    const stack = testStack();
+    const vpc = new ec2.Vpc(stack, 'VPC');
+
+    // WHEN
+    new DatabaseClusterFromSnapshot(stack, 'Database', {
+      engine: DatabaseClusterEngine.aurora({ version: AuroraEngineVersion.VER_1_22_2 }),
+      instanceProps: {
+        vpc,
+      },
+      snapshotIdentifier: 'mySnapshot',
+    });
+
+    // THEN
+    expect(stack).to(haveResource('AWS::RDS::DBCluster', {
+      Properties: {
+        Engine: 'aurora',
+        EngineVersion: '5.6.mysql_aurora.1.22.2',
+        DBSubnetGroupName: { Ref: 'DatabaseSubnets56F17B9A' },
+        VpcSecurityGroupIds: [{ 'Fn::GetAtt': ['DatabaseSecurityGroup5C91FDCB', 'GroupId'] }],
+        SnapshotIdentifier: 'mySnapshot',
+      },
+      DeletionPolicy: ABSENT,
+      UpdateReplacePolicy: 'Snapshot',
+    }, ResourcePart.CompleteDefinition));
+
+    expect(stack).to(countResources('AWS::RDS::DBInstance', 2));
+
+    test.done();
+  },
 };
 
 function testStack() {