Use Case

C++ AWS S3 Client/Service. Use this client to perform S3 select operations on Parquet file stored on AWS S3. Integrates with the Boost Beast HTTP Service and Router and enabled S3 Select over HTTP(s).

Sample Usage of AWS S3 Select: Sravz YTD Page

Session 1:

Perform S3 Select
Perform S3 Select over HTTP

Libraries Used

AWS C++ SDK

Session 1

Dataflow Diagram

Video explanation of the code

Source Code

	#include <aws_s3_service.hpp>
	#include <aws/core/auth/AWSCredentialsProvider.h>
	#include <aws/core/Aws.h>
	#include <aws/core/client/ClientConfiguration.h>
	#include <aws/core/utils/memory/stl/AWSStringStream.h>
	#include <aws/s3/model/Bucket.h>
	#include <aws/s3/model/CSVInput.h>
	#include <aws/s3/model/CSVOutput.h>
	#include <aws/s3/model/DeleteObjectRequest.h>
	#include <aws/s3/model/GetObjectRequest.h>
	#include <aws/s3/model/InputSerialization.h>
	#include <aws/s3/model/JSONInput.h>
	#include <aws/s3/model/JSONOutput.h>
	#include <aws/s3/model/ListObjectsRequest.h>
	#include <aws/s3/model/Object.h>
	#include <aws/s3/model/OutputSerialization.h>
	#include <aws/s3/model/ParquetInput.h>
	#include <aws/s3/model/PutObjectRequest.h>
	#include <aws/s3/model/SelectObjectContentRequest.h>
	#include <aws/s3/S3Client.h>
	#include <boost/log/core.hpp>
	#include <boost/log/expressions.hpp>
	#include <boost/log/trivial.hpp>
	#include <string>
	#include <util.hpp>

	namespace sravz {
	namespace services {
	namespace aws {
	namespace s3 {
	S3Client::S3Client(const std::string& region, const std::string& bucket, Aws::SDKOptions options, S3ClientType type)
	:m_region(region),
	m_bucket(bucket),
	options_(options),
	type_(type)
	{
	Aws::Client::ClientConfiguration clientConfig;
	switch (type)
	{
	case S3ClientType::CONTABO:
	{
	clientConfig.endpointOverride = Aws::String("usc1.contabostorage.com");
	clientConfig.scheme = Aws::Http::Scheme::HTTPS;
	std::string CONTABO_KEY;
	std::string CONTABO_SECRET;
	getenv("CONTABO_KEY", CONTABO_KEY);
	getenv("CONTABO_SECRET", CONTABO_SECRET);
	Aws::Auth::AWSCredentials credentialsContabo(CONTABO_KEY, CONTABO_SECRET);
	m_s3Client = Aws::S3::S3Client(credentialsContabo, clientConfig, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false);
	break;
	}
	case S3ClientType::AWS:
	{
	std::string AWS_ACCESS_KEY_ID;
	std::string AWS_SECRET_ACCESS_KEY;
	getenv("AWS_ACCESS_KEY_ID", AWS_ACCESS_KEY_ID);
	getenv("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY);
	Aws::Auth::AWSCredentials credentialsAWS(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY);
	m_s3Client = Aws::S3::S3Client(credentialsAWS);
	break;
	}
	default:
	std::cout << "Unsupported message type" << std::endl;
	break;
	}
	}

	bool S3Client::putObject(const std::string& key, const std::string& value)
	{
	Aws::S3::Model::PutObjectRequest request;
	request.SetBucket(m_bucket);
	request.SetKey(key);
	// Creates a shared pointer of Aws::StringStream
	auto stream = Aws::MakeShared<Aws::StringStream>("putObject", "");
	*stream << value;
	request.SetBody(stream);

	auto outcome = m_s3Client.PutObject(request);
	if (outcome.IsSuccess()) {
	return true;
	} else {
	std::cout << "Failed with error: " << outcome.GetError() << std::endl;
	return false;
	}
	}

	std::string S3Client::getObject(const std::string& key)
	{
	Aws::S3::Model::GetObjectRequest request;
	request.SetBucket(m_bucket);
	request.SetKey(key);

	auto outcome = m_s3Client.GetObject(request);
	if (outcome.IsSuccess()) {
	// Takes ownership of the GetObjectResult
	auto& retrievedObject = outcome.GetResultWithOwnership().GetBody();
	std::stringstream ss;
	ss << retrievedObject.rdbuf();
	return ss.str();
	} else {
	std::cout << "Failed with error: " << outcome.GetError() << std::endl;
	return "";
	}
	}

	std::vector<std::string> S3Client::listObjects()
	{
	Aws::S3::Model::ListObjectsRequest request;
	request.SetBucket(m_bucket);

	std::vector<std::string> objectKeys;
	auto outcome = m_s3Client.ListObjects(request);
	if (outcome.IsSuccess()) {
	auto objects = outcome.GetResult().GetContents();
	for (const auto& object : objects) {
	objectKeys.push_back(object.GetKey());
	}
	} else {
	std::cout << "Failed with error: " << outcome.GetError() << std::endl;
	}
	return objectKeys;
	}

	bool S3Client::deleteObject(const std::string& key)
	{
	Aws::S3::Model::DeleteObjectRequest request;
	request.SetBucket(m_bucket);
	request.SetKey(key);

	auto outcome = m_s3Client.DeleteObject(request);
	if (outcome.IsSuccess()) {
	return true;
	} else {
	std::cout << "Failed with error: " << outcome.GetError() << std::endl;
	return false;
	}
	}

	std::string S3Client::s3select(const std::string& bucket, const std::string& key, const std::string& column, const std::string& in_clause)
	{
	Aws::S3::Model::SelectObjectContentRequest request;
	request.SetBucket(bucket);
	request.SetKey(key);
	request.SetExpressionType(Aws::S3::Model::ExpressionType::SQL);
	std::ostringstream buffer;
	buffer << "[";

	if (!in_clause.empty()) {
	std::vector<std::string> params;
	std::stringstream ss(in_clause);
	std::string param;
	while (std::getline(ss, param, ',')) {
	params.push_back(param);
	}
	std::string inClause = "(";
	for (auto it = params.begin(); it != params.end(); ++it)
	{
	inClause += "'" + *it + "'";
	if (it != params.end() - 1)
	inClause += ",";
	}
	inClause += ")";
	BOOST_LOG_TRIVIAL(info) << "Select statement: SELECT * FROM S3Object WHERE column IN " << inClause;
	request.SetExpression("SELECT * FROM S3Object WHERE " + column + " IN " + inClause);
	} else {
	request.SetExpression("SELECT * FROM S3Object");
	}


	// Set the input and output serialization formats for Parquet
	Aws::S3::Model::InputSerialization input_serialization;
	input_serialization.SetParquet(Aws::S3::Model::ParquetInput());
	Aws::S3::Model::OutputSerialization output_serialization;
	Aws::S3::Model::JSONOutput jsonOutput;
	jsonOutput.SetRecordDelimiter(",");
	output_serialization.SetJSON(jsonOutput);
	request.SetInputSerialization(input_serialization);
	request.SetOutputSerialization(output_serialization);

	Aws::S3::Model::SelectObjectContentHandler handler;
	handler.SetRecordsEventCallback([&](const Aws::S3::Model::RecordsEvent& recordsEvent)
	{
	auto recordsVector = recordsEvent.GetPayload();
	Aws::String records(recordsVector.begin(), recordsVector.end());
	buffer << records.c_str();
	});
	handler.SetStatsEventCallback([&](const Aws::S3::Model::StatsEvent& statsEvent)
	{

	});

	// Register a callback function to handle the S3 Select events
	request.SetEventStreamHandler(handler);

	// Issue the S3 Select request
	auto outcome = m_s3Client.SelectObjectContent(request);
	if (outcome.IsSuccess())
	{
	std::cout << "S3 Select request succeeded" << std::endl;
	}
	else
	{
	std::cout << "S3 Select request failed: " << outcome.GetError().GetMessage() << std::endl;
	}
	auto outputString = buffer.str();
	if (!outputString.empty() && outputString.back() == ',') {
	outputString.pop_back();
	}
	outputString += "]";
	return outputString;
	}

	S3Client::~S3Client()
	{
	// Aws::ShutdownAPI(options_);
	}
	}
	}
	}
	}

view raw aws_s3_service.cpp hosted with ❤ by GitHub

	#pragma once
	#include <string>
	#include <vector>
	#include <aws/core/Aws.h>
	#include <aws/s3/S3Client.h>

	namespace sravz {
	namespace services {
	namespace aws {
	namespace s3 {
	enum class S3ClientType {
	CONTABO = 1,
	AWS = 2
	};

	class S3Client {
	public:
	S3Client(const std::string& region, const std::string& bucket, Aws::SDKOptions options, S3ClientType type);
	bool putObject(const std::string& key, const std::string& value);
	std::string getObject(const std::string& key);
	std::vector<std::string> listObjects();
	bool deleteObject(const std::string& key);
	std::string s3select(const std::string& bucket, const std::string& key, const std::string& column, const std::string& value);
	~S3Client();
	private:
	std::string m_region;
	std::string m_bucket;
	Aws::Client::ClientConfiguration m_clientConfig;
	Aws::S3::S3Client m_s3Client;
	Aws::SDKOptions options_;
	S3ClientType type_;
	// void processRecordsEvent_(const Aws::S3::Model::SelectObjectContentEventStream& eventStream);
	};
	}
	}
	}
	}

view raw aws_s3_service.hpp hosted with ❤ by GitHub

	/**
	* SRAVZ LLC
	**/

	#include "main.hpp"
	#include <boost/version.hpp>
	#include <aws/core/Aws.h>
	using namespace std;

	int main(int ac, char* av[])
	{
	try {
	int opt;
	string config_file;
	string sub_program_name;
	string redis_host;
	string redis_port;
	string redis_password;

	set<string> sub_program_options{ "ws-vendor", "redis-uploader", "webserver" };
	set<string> log_levels{ "debug", "info", "warning", "error" };
	// Declare a group of options that will be
	// allowed only on command line
	po::options_description generic("Generic options");
	generic.add_options()
	("version,v", "print version string")
	("help", "produce help message")
	("config,c", po::value<string>(&config_file)->default_value("app.cfg"),
	"name of a file of a configuration.")
	("sub-program-name,s", po::value<string>(&sub_program_name)->default_value("ws-vendor"),
	"Module name to run. Options: ws-vendor\|redis-uploader\|webserver.")
	;

	// Declare a group of options that will be
	// allowed both on command line and in
	// config file
	po::options_description config("Configuration");
	config.add_options()
	("ws-vendor-hostname",
	po::value<string>(),
	"websocket vendor hostname")
	("ws-vendor-port",
	po::value<string>(),
	"websocket vendor port")
	("ws-vendor-greeting-str",
	po::value<string>(),
	"websocket vendor greeting string")
	("ws-vendor-endpoint",
	po::value<string>(),
	"websocket vendor endpoint")
	("ws-vendor-no-threads",
	po::value<int>(),
	"websocket vendor no of threads")
	("redis-topics",
	po::value<string>(),
	"comma separated list of redis topics")
	("webserver-hostname",
	po::value<string>(),
	"webserver hostname or ip address to run on")
	("webserver-port",
	po::value<string>(),
	"webserver port")
	("webserver-root-dir",
	po::value<string>(),
	"webserver root dir")
	("webserver-no-threads",
	po::value<string>(),
	"webserver no of threads")
	("webserver-topics",
	po::value<string>(),
	"comma separated list of topics")
	("log-level",
	po::value<string>(),
	"log level")
	;


	// Hidden options, will be allowed both on command line and
	// in config file, but will not be shown to the user.
	po::options_description hidden("Hidden options");
	// hidden.add_options()
	// ("input-file", po::value< vector<string> >(), "input file")
	// ;


	po::options_description cmdline_options;
	cmdline_options.add(generic).add(config).add(hidden);

	po::options_description config_file_options;
	config_file_options.add(config).add(hidden);

	po::options_description visible("Allowed options");
	visible.add(generic).add(config);

	po::positional_options_description p;
	// p.add("input-file", -1);

	// Calls to store, parse_command_line and notify functions
	// cause vm to contain all the options found on the command line.
	po::variables_map vm;
	store(po::command_line_parser(ac, av).
	options(cmdline_options).positional(p).run(), vm);
	notify(vm);

	ifstream ifs(config_file.c_str());
	if (!ifs)
	{
	BOOST_LOG_TRIVIAL(info) << "can not open config file: " << config_file;
	return 0;
	}
	else
	{
	store(parse_config_file(ifs, config_file_options), vm);
	notify(vm);
	}

	// If variables map contains help. Please help
	if (vm.count("help")) {
	BOOST_LOG_TRIVIAL(info) << visible;
	return 0;
	}

	if (vm.count("version")) {
	BOOST_LOG_TRIVIAL(info) << "backend-cpp, version 1.0";
	return 0;
	}

	// If variables map contains sub_program_name. Check it is a valid sub_program.
	if (sub_program_options.find(sub_program_name) == sub_program_options.end()) {
	BOOST_LOG_TRIVIAL(info) << "sub-program-name: " << sub_program_name << " invalid. should be one of: ";
	for (string const& sub_program : sub_program_options)
	{
	BOOST_LOG_TRIVIAL(info) << sub_program << ' ';
	}
	return 1;
	}

	std::string hostname;
	boost::program_options::options_description desc_env;
	// TODO: Add these back if we need to pass the parameters at the cli
	desc_env.add_options()("redis_host", boost::program_options::value<std::string>(&redis_host));
	desc_env.add_options()("redis_port", boost::program_options::value<std::string>(&redis_port));
	desc_env.add_options()("redis_password", boost::program_options::value<std::string>(&redis_password));

	boost::program_options::variables_map vm_env;
	boost::program_options::store(boost::program_options::parse_environment(desc_env,
	[](const std::string& i_env_var)
	{

	switch(get_hash(i_env_var)) {
	case "REDIS_HOST"_ :
	{
	return "redis_host";
	break;
	}
	case "REDIS_PORT"_ :
	return "redis_port";
	break;
	case "REDIS_PASSWORD"_ :
	return "redis_password";
	break;
	default:
	return "";
	}
	}),
	vm_env);
	boost::program_options::notify(vm_env);
	// BOOST_LOG_TRIVIAL(info) << "Environment varaibles: \n";
	// BOOST_LOG_TRIVIAL(info) << "redis_host: " << vm_env["redis_host"].as<string>() << '\n';
	// BOOST_LOG_TRIVIAL(info) << "redis_port: " << vm_env["redis_port"].as<string>() << '\n';
	// BOOST_LOG_TRIVIAL(info) << "redis_password: " << vm_env["redis_password"].as<string>() << '\n';

	// Init program

	switch(get_hash(vm["log-level"].as<string>())) {
	case "debug"_ :
	{
	logging::core::get()->set_filter
	(
	logging::trivial::severity >= logging::trivial::debug
	);
	break;
	}
	case "info"_ :
	logging::core::get()->set_filter
	(
	logging::trivial::severity >= logging::trivial::info
	);
	break;
	case "warning"_ :
	logging::core::get()->set_filter
	(
	logging::trivial::severity >= logging::trivial::warning
	);
	break;
	case "error"_ :
	logging::core::get()->set_filter
	(
	logging::trivial::severity >= logging::trivial::error
	);
	break;
	default:
	BOOST_LOG_TRIVIAL(info) << "Unsupported log level: " << vm["log-level"].as<string>() << endl;
	return 1;
	}

	BOOST_LOG_TRIVIAL(info) << "Boost version: " << BOOST_VERSION;
	BOOST_LOG_TRIVIAL(info) << "sub_program_name";
	switch(get_hash(sub_program_name)) {
	case "ws-vendor"_ :
	{
	ws_vendor(vm["ws-vendor-hostname"].as<string>(),
	vm["ws-vendor-port"].as<string>(),
	vm["ws-vendor-greeting-str"].as<string>(),
	vm["ws-vendor-endpoint"].as<string>(),
	vm["ws-vendor-no-threads"].as<int>());
	break;
	}
	case "redis-uploader"_ :
	redis_uploader(vm["redis-topics"].as<string>());
	break;
	case "webserver"_ : {
	Aws::SDKOptions options;
	Aws::InitAPI(options);
	webserver(vm["webserver-hostname"].as<string>(),
	vm["webserver-port"].as<string>(),
	vm["webserver-root-dir"].as<string>(),
	vm["webserver-no-threads"].as<string>(),
	vm["webserver-topics"].as<string>());
	// Shutdown the AWS SDK for C++ when done
	Aws::ShutdownAPI(options);
	break;
	}
	default:
	BOOST_LOG_TRIVIAL(info) << "Unsupported option: " << sub_program_name << endl;
	}

	}
	catch(exception& e)
	{
	BOOST_LOG_TRIVIAL(info) << e.what();
	return 1;
	}
	return 0;
	}

view raw main.cpp hosted with ❤ by GitHub

	#include <cstdio>
	#include <iostream>
	#include <memory>
	#include <router.hpp>
	#include <sstream>
	#include <string>
	#include <vector>
	#include <boost/log/core.hpp>
	#include <boost/log/expressions.hpp>
	#include <boost/log/trivial.hpp>

	using namespace boost::urls;

	namespace sravz {
	void router::init_()
	{
	// Initialize services
	mongoClient_ = std::make_shared<sravz::services::mongo::MongoClient>(std::move("mongodb://sravz:sravz@mongo:27017/sravz"));
	Aws::SDKOptions options;
	s3Client_ = std::make_shared<sravz::services::aws::s3::S3Client>("us-east-1", "sravz-data", std::move(options), sravz::services::aws::s3::S3ClientType::AWS);
	// Initialize controllers
	mongoController_ = std::make_shared<sravz::controllers::mongo::MongoController>(mongoClient_);
	awsS3Controller_ = std::make_shared<sravz::controllers::aws::AWSS3Controller>(s3Client_);
	}

	std::pair<boost::beast::http::status, std::string> router::route(boost::beast::http::verb verb, const std::string& uri, const std::string& body)
	{
	try {
	BOOST_LOG_TRIVIAL(info) << "Request URL: " << uri;
	//result<boost::urls::url_view> r = boost::urls::parse_uri(uri);
	//url_view u = r.value();
	boost::urls::url_view uv(uri);
	auto segments = uv.segments();
	if (!segments.size())
	return { boost::beast::http::status::not_implemented, "Unsupported path" };
	// return "Path Not Supported";
	boost::json::object obj;
	std::map<std::string, std::string> params_map;
	for (auto param: uv.params()) {
	obj[param.key] = param.value;
	params_map[param.key] = param.value;
	}
	auto params_json = boost::json::serialize(obj);
	switch (verb) {
	case boost::beast::http::verb::get:
	if (*segments.begin() == "mdb")
	{
	if (!uv.params().size()) {
	// Avoid sending the complete collection
	return { boost::beast::http::status::bad_request, "Params required for GET method" };
	}
	auto it = std::next(segments.begin());
	if (it != segments.end()) {
	return { boost::beast::http::status::ok, mongoController_->get(*it, params_json) };
	}
	} else if (*segments.begin() == "s3") {
	if (!uv.params().size()) {
	// Avoid sending the complete collection
	return { boost::beast::http::status::bad_request, "Params required for GET method" };
	}
	if (!obj.contains("bucket") \|\| !obj.contains("key") \|\| !obj.contains("column") \|\| !obj.contains("in_clause")) {
	return { boost::beast::http::status::bad_request, "bucket, key, col and in_clause params required" };
	}
	BOOST_LOG_TRIVIAL(info) << "S3-Select: Bucket " << params_map["bucket"] << " Key: " << params_map["key"] << " Column: " << params_map["column"] << " in_clause: " << params_map["in_clause"];
	return {boost::beast::http::status::ok,
	awsS3Controller_->get(params_map["bucket"], params_map["key"], params_map["column"], params_map["in_clause"])
	};
	}
	return { boost::beast::http::status::not_implemented, "Unsupported service" };
	break;
	case boost::beast::http::verb::post:
	if (*segments.begin() == "mdb")
	{
	auto it = std::next(segments.begin());
	if (it != segments.end()) {
	return { boost::beast::http::status::created, mongoController_->post(*it, body)};
	}
	}
	return { boost::beast::http::status::not_implemented, "Unsupported service" };
	break;
	case boost::beast::http::verb::put:
	if (*segments.begin() == "mdb")
	{
	auto it = std::next(segments.begin());
	if (it != segments.end()) {
	boost::json::value json = boost::json::parse(body);
	std::string filter_json = boost::json::serialize(json.at("filter").as_object());
	std::string update_json = boost::json::serialize(json.at("update").as_object());
	bool update_one = json.at("update_one").as_bool();
	return { boost::beast::http::status::created, mongoController_->put(*it, filter_json, update_json, update_one)};
	}
	}
	return { boost::beast::http::status::not_implemented, "Unsupported service" };
	break;
	case boost::beast::http::verb::delete_:
	if (*segments.begin() == "mdb")
	{
	auto it = std::next(segments.begin());
	if (it != segments.end()) {
	boost::json::value json = boost::json::parse(body);
	std::string filter_json = boost::json::serialize(json.at("filter").as_object());
	bool delete_one = json.at("delete_one").as_bool();
	return { boost::beast::http::status::ok, mongoController_->_delete(*it, filter_json, delete_one)};
	}
	}
	return { boost::beast::http::status::not_implemented, "Unsupported service" };
	break;
	default:
	return { boost::beast::http::status::method_not_allowed, "Unsupported method" };
	break;
	}
	return { boost::beast::http::status::method_not_allowed, "Unsupported method" };
	}
	catch(std::exception& ex)
	{
	return { boost::beast::http::status::internal_server_error, "Error" };
	BOOST_LOG_TRIVIAL(error) << "Router error" << ex.what();
	}
	return { boost::beast::http::status::not_implemented, "Unsupported service" };
	}
	}

view raw router.cpp hosted with ❤ by GitHub

	#pragma once
	// Service includes
	#include <mongo_service.hpp>
	#include <aws_s3_service.hpp>
	// Controller includes
	#include <mongo_controller.hpp>
	#include <aws_s3_controller.hpp>
	#include <boost/beast/http.hpp>
	#include <boost/json.hpp>
	#include <boost/url.hpp>


	namespace sravz {
	class router
	{
	public:
	router()
	{
	init_();
	}
	std::pair<boost::beast::http::status, std::string> route(boost::beast::http::verb method, const std::string& uri, const std::string& body);
	private:
	void init_();
	// Services
	std::shared_ptr<sravz::services::mongo::MongoClient> mongoClient_;
	std::shared_ptr<sravz::services::aws::s3::S3Client> s3Client_;
	// Controllers
	std::shared_ptr<sravz::controllers::mongo::MongoController> mongoController_;
	std::shared_ptr<sravz::controllers::aws::AWSS3Controller> awsS3Controller_;
	};
	}

view raw router.hpp hosted with ❤ by GitHub