private static void createDeliveryStream ( ) throws Exception { boolean deliveryStreamExists = false ; LOG . info ( "Checking if " + deliveryStreamName + " already exits" ) ; List < String > deliveryStreamNames = listDeliveryStreams ( ) ; if ( deliveryStreamNames != null && deliveryStreamNames . contains ( deliveryStreamName ) ) { deliveryStreamExists = true ; LOG . info ( "DeliveryStream " + deliveryStreamName + " already exists. Not creating the new delivery stream" ) ; } else { LOG . info ( "DeliveryStream " + deliveryStreamName + " does not exist" ) ; } if ( ! deliveryStreamExists ) { // Create DeliveryStream CreateDeliveryStreamRequest createDeliveryStreamRequest = new CreateDeliveryStreamRequest ( ) ; createDeliveryStreamRequest . setDeliveryStreamName ( deliveryStreamName ) ; S3DestinationConfiguration redshiftS3Configuration = new S3DestinationConfiguration ( ) ; redshiftS3Configuration . setBucketARN ( s3BucketARN ) ; redshiftS3Configuration . setPrefix ( s3ObjectPrefix ) ; BufferingHints bufferingHints = null ; if ( s3DestinationSizeInMBs != null || s3DestinationIntervalInSeconds != null ) { bufferingHints = new BufferingHints ( ) ; bufferingHints . setSizeInMBs ( s3DestinationSizeInMBs ) ; bufferingHints . setIntervalInSeconds ( s3DestinationIntervalInSeconds ) ; } redshiftS3Configuration . setBufferingHints ( bufferingHints ) ; // Create and set IAM role so that firehose service has access to the S3Buckets to put data. // Please check the trustPolicyDocument.json and permissionsPolicyDocument.json files // for the trust and permissions policies set for the role. String iamRoleArn = createIamRole ( s3ObjectPrefix ) ; redshiftS3Configuration . setRoleARN ( iamRoleArn ) ; CopyCommand copyCommand = new CopyCommand ( ) ; copyCommand . withCopyOptions ( copyOptions ) . withDataTableName ( dataTableName ) ; RedshiftDestinationConfiguration redshiftDestinationConfiguration = new RedshiftDestinationConfiguration ( ) ; redshiftDestinationConfiguration . withClusterJDBCURL ( clusterJDBCUrl ) . withRoleARN ( iamRoleArn ) . withUsername ( username ) . withPassword ( password ) . withCopyCommand ( copyCommand ) . withS3Configuration ( redshiftS3Configuration ) ; createDeliveryStreamRequest . setRedshiftDestinationConfiguration ( redshiftDestinationConfiguration ) ; firehoseClient . createDeliveryStream ( createDeliveryStreamRequest ) ; // The Delivery Stream is now being created. LOG . info ( "Creating DeliveryStream : " + deliveryStreamName ) ; waitForDeliveryStreamToBecomeAvailable ( deliveryStreamName ) ; } }
Method to create delivery stream with Redshift destination configuration .