S3 Bucket Analysis

CREATE EXTERNAL TABLE prod_inventory(
  `bucket` string,
  key string,
  version_id string,
  is_latest boolean,
  is_delete_marker boolean,
  size bigint,
  last_modified_date timestamp,
  e_tag string,
  is_multipart_uploaded boolean
  )
  PARTITIONED BY (dt string)
  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
  STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
  OUTPUTFORMAT  'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
  LOCATION 's3://prod.inventory.sagebase.org/inventory/proddata.sagebase.org/defaultInventory/hive';

MSCK REPAIR TABLE prod_inventory

SELECT dt, COUNT(*), SUM(size) FROM prod_inventory GROUP BY dt

SELECT dt, COUNT(*), SUM(size) FROM prod_inventory WHERE is_multipart_uploaded = true 
GROUP BY dt

CREATE TABLE FILES_PROD AS 
SELECT *, ROW_NUMBER() OVER(PARTITION BY BUCKET_NAME, `KEY` ORDER BY ID) AS D_ID 
FROM FILES WHERE BUCKET_NAME = 'proddata.sagebase.org'

SELECT COUNT(*) AS TOTAL_COUNT, SUM(CONTENT_SIZE) AS TOTAL_SIZE FROM FILES_PROD

SELECT COUNT(*) AS TOTAL_COUNT, SUM(CONTENT_SIZE) AS TOTAL_SIZE FROM FILES_PROD WHERE D_ID = 1

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

connection_mysql8 = {
    "url": "jdbc:mysql://restored-prod-332-db-20201105.c5sxx7pot9i8.us-east-1.rds.amazonaws.com:3306/prod332",
    "dbtable": "FILES_PROD",
    "user": "prod332user",
    "password": "secret",
    "customJdbcDriverS3Path": "s3://analytics.sagebase.org/mysql/mysql-connector-java-8.0.22.jar",
    "customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver",
    "hashexpression": "D_ID = 1 AND ID"
}

## @type: DataSource
## @args: [connection_type = "mysql", connection_options=connection_mysql8, transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql8, transformation_ctx = "datasource0")

## @type: ApplyMapping
## @args: [mapping = [("ID", "BIGINT", "id", "BIGINT"), ("CREATED_ON", "timestamp", "CREATED_ON", "timestamp"), ("CREATED_BY", "BIGINT", "CREATED_BY", "BIGINT"), ("KEY", "string", "key", "string"), ("CONTENT_SIZE", "BIGINT", "CONTENT_SIZE", "BIGINT"), ("CONTENT_MD5", "string", "CONTENT_MD5", "string"), ("IS_PREVIEW", "boolean", "IS_PREVIEW", "boolean"), ("PREVIEW_ID", "BIGINT", "PREVIEW_ID", "BIGINT"), transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("ID", "BIGINT", "id", "BIGINT"), ("CREATED_ON", "timestamp", "CREATED_ON", "timestamp"), ("CREATED_BY", "BIGINT", "CREATED_BY", "BIGINT"), ("KEY", "string", "key", "string"), ("CONTENT_SIZE", "BIGINT", "CONTENT_SIZE", "BIGINT"), ("CONTENT_MD5", "string", "CONTENT_MD5", "string"), ("IS_PREVIEW", "boolean", "IS_PREVIEW", "boolean"), ("PREVIEW_ID", "BIGINT", "PREVIEW_ID", "BIGINT")], transformation_ctx = "applymapping1")
## @type: ResolveChoice
## @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"]
## @return: resolvechoice2
## @inputs: [frame = applymapping1]
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
## @type: DropNullFields
## @args: [transformation_ctx = "dropnullfields3"]
## @return: dropnullfields3
## @inputs: [frame = resolvechoice2]
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
## @type: DataSink
## @args: [connection_type = "s3", connection_options = {"path": "s3://analytics.sagebase.org/prod332/file_handles"}, format = "parquet", transformation_ctx = "datasink4"]
## @return: datasink4
## @inputs: [frame = dropnullfields3]
datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://analytics.sagebase.org/prod332/file_handles"}, format = "parquet", transformation_ctx = "datasink4")
job.commit()

Name	prod332FileHandlesCrawler
Description	
Create a single schema for each S3 path	false
Security configuration	
Tags	
-
State	Ready
Schedule	
Last updated	Fri Nov 13 12:13:21 GMT-800 2020
Date created	Fri Nov 13 10:40:11 GMT-800 2020
Database	files_inventory
Service role	AWSGlueServiceRoleDefault
Selected classifiers	
Data store	S3
Include path	s3://analytics.sagebase.org/prod332/file_handles
Connection	
Exclude patterns	
Configuration options
Schema updates in the data store	Update the table definition in the data catalog.
Object deletion in the data store	Mark the table as deprecated in the data catalog.

SELECT COUNT(*) AS MISSING_COUNT, SUM(F.CONTENT_SIZE) AS MISSING_SIZE 
FROM file_handles F LEFT JOIN prod_inventory I ON F.KEY = I.KEY
WHERE I.KEY IS NULL

SELECT COUNT(*) AS MISSING_COUNT, SUM(I.SIZE) AS MISSING_SIZE 
FROM prod_inventory I LEFT JOIN file_handles F ON I.KEY = F.KEY
WHERE I.dt = '2020-11-05-00-00' AND F.KEY IS NULL

package org.sagebionetworks.project;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Date;

import org.sagebionetworks.aws.MultiFactorAuthenticationCredentialProvider;

import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.ListMultipartUploadsRequest;
import com.amazonaws.services.s3.model.MultipartUploadListing;

public class MultipartUploadsScanner {
	
	private AmazonS3 s3Client;
	
	public static void main(String[] args) {

		String mfaDeviceArn = "mfaDeviceArn";
		
		AmazonS3 s3Client = buildS3Client(mfaDeviceArn);
		
		String bucket = "proddata.sagebase.org";
		
		new MultipartUploadsScanner(s3Client).scan(bucket);
	}
		
	private static AmazonS3 buildS3Client(String mfaDeviceArn) {	
		AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
		builder.withCredentials(new MultiFactorAuthenticationCredentialProvider(mfaDeviceArn));
		builder.withRegion(Regions.US_EAST_1);
		return builder.build();
	}

	
	public MultipartUploadsScanner(AmazonS3 s3Client) {
		this.s3Client = s3Client;
	}
	
	void scan(String bucket) {
		
		ListMultipartUploadsRequest request = new ListMultipartUploadsRequest(bucket);
		
		boolean hasNext = true;
		String keyMarker = null;
		String uploadIdMarker = null;
		
		int totalCount = 0;
		int totalOldCount = 0;
		
		Date oneMonthAgo = Date.from(Instant.now().minus(30, ChronoUnit.DAYS));
		
		while (hasNext) {
			
			request.setKeyMarker(keyMarker);
			request.setUploadIdMarker(uploadIdMarker);
			
			MultipartUploadListing result = s3Client.listMultipartUploads(request);
			
			int count = result.getMultipartUploads().size();
			long oldCount = result.getMultipartUploads().stream().filter( u-> u.getInitiated().before(oneMonthAgo)).count();
			
			totalCount += count;
			totalOldCount += oldCount;
			
			System.out.println(String.format("Batch Count: %s, Old Count: %s (Total: %s, Total Old: %s)", count, oldCount, totalCount, totalOldCount));
			
			if (result.isTruncated()) {
				keyMarker = result.getNextKeyMarker();
				uploadIdMarker = result.getNextUploadIdMarker();
				System.out.println(String.format("Has next: %s, %s", uploadIdMarker, keyMarker));
			} else {
				hasNext = false;
			}
		}
		
		System.out.println(String.format("Number of multipart uploads: %s (Old: %s)", totalCount, totalOldCount));
	}

}

SELECT COUNT(*) FROM MULTIPART_UPLOAD U WHERE U.STATE = 'UPLOADING'