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

WITH 
U_R_F AS (
	SELECT DISTINCT R.FILE_HANDLE_ID FROM JDOREVISION R WHERE R.FILE_HANDLE_ID IS NOT NULL
),
U_F AS (
	SELECT MAX(F.CONTENT_SIZE) AS CONTENT_SIZE FROM U_R_F R JOIN FILES_PROD F ON R.FILE_HANDLE_ID = F.ID GROUP BY F.`KEY`
)
SELECT COUNT(*), SUM(CONTENT_SIZE) FROM U_F

CREATE TABLE `FILES_TABLES` (
	`TABLE_ID` BIGINT(20) NOT NULL,
	`ID` BIGINT(20) NOT NULL
)

package org.sagebionetworks.project;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.sql.DataSource;

import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.commons.lang3.time.StopWatch;
import org.springframework.jdbc.core.JdbcTemplate;

public class SynapseTablesFileHandlesScanner {

	private static final String DB_DRIVER = "com.mysql.cj.jdbc.Driver";
	private static final String DB_URL = "jdbc:mysql://restored-prod-332-table-0-20201105.c5sxx7pot9i8.us-east-1.rds.amazonaws.com:3306/prod332";
	private static final String DB_USER = "prod332user";
	private static final String DB_PASSWORD = "secret";
	private static final String SCHEMA = "prod332";
	
	public static void main(String[] args) {
		DataSource dataSource = getDataSource();
		JdbcTemplate jdbcTemplate = getTemplate(dataSource);
		ExecutorService executorService = Executors.newFixedThreadPool(20);
		ScheduledExecutorService schedulerService = Executors.newScheduledThreadPool(1);
		
		final SynapseTablesFileHandlesScanner scanner = new SynapseTablesFileHandlesScanner(jdbcTemplate);
		
		int limit = 20000;
		
		final List<FileTable> tables = scanner.getFileTables(limit);
		final int size = tables.size();
		final AtomicInteger counter = new AtomicInteger();
		
		final StopWatch watch = StopWatch.createStarted();
		
		System.out.println(String.format("Copying data from %d tables...", size));
		
		tables.forEach(table -> {
			executorService.submit(() -> {
				timedRunnable(() -> {
					int rowNum = scanner.copyFileHandles(table);
					System.out.println(String.format("Finished copying %d rows from table %s", rowNum, table.getName()));
					counter.incrementAndGet();
				}, String.format("Copying table %s (Id: %d, Est Size: %d)", table.getName(), table.getId(), table.getSize()));
			});
			System.out.println(String.format("Table %s submitted.", table.getName()));
		});	
		
		schedulerService.scheduleWithFixedDelay(() -> {
			int currentCount = counter.get();
			try {
				System.out.println(String.format("-----\n%d/%d tables processed (%d%%, Elapsed Time: %d sec)\n-----", currentCount, size, (currentCount * 100 / size), watch.getTime(TimeUnit.SECONDS)));
			} catch (Exception e) {
				e.printStackTrace();
			}
		}, 5, 5, TimeUnit.SECONDS);
				
		terminateExecutorService(executorService, 5, TimeUnit.HOURS);
		terminateExecutorService(schedulerService, 5, TimeUnit.SECONDS);
		
		watch.stop();
		
		System.out.println(String.format("Copying data from %d tables...DONE (Took: %d mins)", size, watch.getTime(TimeUnit.MINUTES)));
	}
	
	private static void terminateExecutorService(ExecutorService executorService, long timeout, TimeUnit unit) {
		executorService.shutdown();
		
		try {
		    if (!executorService.awaitTermination(timeout, unit)) {
		        executorService.shutdownNow();
		    } 
		} catch (InterruptedException e) {
		    executorService.shutdownNow();
		}
	}
	
	private static void timedRunnable(Runnable runnable, String text) {
		StopWatch watch = StopWatch.createStarted();
		System.out.println(String.format("%s...", text));
		try {
			runnable.run();
		} catch (Exception e) {
			System.out.println(String.format("%s...FAILED: %s", text, e.getMessage()));
			e.printStackTrace();
		}
		watch.stop();
		System.out.println(String.format("%s...DONE (Took: %d sec)", text, watch.getTime(TimeUnit.SECONDS)));
	}
	
	private static JdbcTemplate getTemplate(DataSource dataSource) {
		return new JdbcTemplate(dataSource);
	}

	private static DataSource getDataSource() {

		BasicDataSource ds = new BasicDataSource();

		ds.setDriverClassName(DB_DRIVER);
		ds.setUsername(DB_USER);
		ds.setPassword(DB_PASSWORD);
		ds.setUrl(DB_URL);
		ds.setMinIdle(5);
		ds.setMaxIdle(50);
		ds.setMaxTotal(50);

		return ds;
	}
	
	private int copyFileHandles(FileTable table) {
		return jdbcTemplate.update("INSERT INTO FILES_TABLES SELECT " + table.getId() + ", FILE_ID FROM " + table.getName());
	}

	private JdbcTemplate jdbcTemplate;
	
	public SynapseTablesFileHandlesScanner(JdbcTemplate jdbcTemplate) {
		this.jdbcTemplate = jdbcTemplate;
	}

	public List<FileTable> getFileTables(int limit) {
		StringBuilder sqlBuilder = new StringBuilder()
				.append("SELECT TABLE_NAME, TABLE_ROWS FROM information_schema.TABLES T WHERE T.TABLE_SCHEMA = \"").append(SCHEMA).append("\"")
				.append(" AND T.TABLE_NAME LIKE 'T%F'")
				.append(" ORDER BY T.TABLE_NAME LIMIT ").append(limit);
		
		return jdbcTemplate.query(sqlBuilder.toString(), (rs, rNum) -> new FileTable(rs.getString(1), rs.getLong(2)));
	}

	private static class FileTable {

		private static final Pattern FILES_TABLE_PATTERN = Pattern.compile("T(\\d+)(_\\d+)?F");
		
		private Long id;
		private String name;
		private Long size;

		public FileTable(String name, Long size) {
			this.id = parseId(name);
			this.name = name;
			this.size = size;
		}
				
		private Long getId() {
			return id;
		}

		public String getName() {
			return name;
		}

		public Long getSize() {
			return size;
		}
				
		private static Long parseId(String name) {
			Matcher matcher = FILES_TABLE_PATTERN.matcher(name);
			if (matcher.matches()) {
				return Long.valueOf(matcher.group(1));
			}
			throw new IllegalArgumentException("Invalid table name " + name);
		}

	}
}

CREATE TABLE `FILES_TABLE_DISTINCT` (
	`ID` BIGINT(20) NOT NULL,
	`REUSED` TINYINT(4) NOT NULL DEFAULT '0'
)

INSERT INTO FILES_TABLE_DISTINCT SELECT DISTINCT F.ID, 0 FROM FILES_TABLES F

SELECT COUNT(DISTINCT F.ID) FROM FILES_TABLE_DISTINCT F JOIN OBJECT_REPLICATION R ON F.ID = R.FILE_ID

UPDATE FILES_TABLE_DISTINCT F JOIN OBJECT_REPLICATION R ON F.ID = R.FILE_ID SET REUSED = TRUE

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-table-0-20201105.c5sxx7pot9i8.us-east-1.rds.amazonaws.com:3306/prod332",
    "dbtable": "FILES_TABLE_DISTINCT",
    "user": "prod332user",
    "password": "secret",
    "customJdbcDriverS3Path": "s3://analytics.sagebase.org/mysql/mysql-connector-java-8.0.22.jar",
    "customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver",
    "hashexpression": "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"), ("REUSED", "TINYINT ", "reused", "TINYINT")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("ID", "BIGINT", "id", "BIGINT"), ("REUSED", "TINYINT", "reused", "TINYINT")], transformation_ctx = "applymapping1")
## @type: ResolveChoice
## @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"]
## @return: resolvechoice2
## @inputs: [frame = applymapping1]
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, specs = [("reused", "cast:boolean")], transformation_ctx = "resolvechoice2")
## @type: DataSink
## @args: [connection_type = "s3", connection_options = {"path": "s3://analytics.sagebase.org/prod332/tables_file_handles"}, format = "parquet", transformation_ctx = "datasink4"]
## @return: datasink4
## @inputs: [frame = dropnullfields3]
datasink4 = glueContext.write_dynamic_frame.from_options(frame = resolvechoice2, connection_type = "s3", connection_options = {"path": "s3://analytics.sagebase.org/prod332/tables_file_handles"}, format = "parquet", transformation_ctx = "datasink4")

job.commit()

SELECT COUNT(*), SUM(F.CONTENT_SIZE) FROM tables_file_handles T JOIN file_handles_with_d_id F ON T.ID = F.ID WHERE T.reused = FALSE

SELECT COUNT(*), SUM(F.CONTENT_SIZE) FROM tables_file_handles T JOIN file_handles_with_d_id F ON T.ID = F.ID WHERE T.reused = TRUE

SELECT COUNT(*) FROM tables_file_handles T JOIN file_handles_with_d_id F ON T.ID = F.ID

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'