You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2019/06/12 15:23:09 UTC
[kudu] branch master updated (53ea7c6 -> 9e3a9fb)
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 53ea7c6 thirdparty: fix build_curl some more
new d25c6f1 docs: Add simplest possible Spark SQL example
new 9e3a9fb KUDU-2848: fetch HMS DB UUID in the background
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/developing.adoc | 37 ++++++++++++++++++++++--------
src/kudu/hms/hms_catalog-test.cc | 24 ++++++++++++++++++++
src/kudu/hms/hms_catalog.cc | 49 +++++++++++++++++++++++++++-------------
src/kudu/hms/hms_catalog.h | 28 ++++++++++++++++++++++-
4 files changed, 112 insertions(+), 26 deletions(-)
[kudu] 02/02: KUDU-2848: fetch HMS DB UUID in the background
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 9e3a9fb1fc0a9a85bb1c948edb8a6b37f98ba2d0
Author: Adar Dembo <ad...@cloudera.com>
AuthorDate: Wed Jun 12 00:03:18 2019 -0700
KUDU-2848: fetch HMS DB UUID in the background
To avoid having a hard startup dependency from Kudu to HMS, let's thunk the
work of fetching the DB UUID to a helper thread. It's a simple approach, but
admittedly I haven't given much thought to whether it preserves reasonable
semantics for clients.
The new test isn't useful because our HMS test artifact doesn't support DB
UUIDs. I'll make sure this gets proper testing after merging though.
Change-Id: I5d190d317b3ee3e69493638e866949cef8fb5d46
Reviewed-on: http://gerrit.cloudera.org:8080/13594
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
---
src/kudu/hms/hms_catalog-test.cc | 24 ++++++++++++++++++++
src/kudu/hms/hms_catalog.cc | 49 +++++++++++++++++++++++++++-------------
src/kudu/hms/hms_catalog.h | 28 ++++++++++++++++++++++-
3 files changed, 84 insertions(+), 17 deletions(-)
diff --git a/src/kudu/hms/hms_catalog-test.cc b/src/kudu/hms/hms_catalog-test.cc
index a8cc0a9..cd32622 100644
--- a/src/kudu/hms/hms_catalog-test.cc
+++ b/src/kudu/hms/hms_catalog-test.cc
@@ -453,6 +453,7 @@ TEST_F(HmsCatalogTest, TestReconnect) {
}
TEST_F(HmsCatalogTest, TestMetastoreUuid) {
+ // Test that if the HMS has a DB UUID, we can retrieve it.
string uuid;
Status s = hms_catalog_->GetUuid(&uuid);
if (s.ok()) {
@@ -460,6 +461,29 @@ TEST_F(HmsCatalogTest, TestMetastoreUuid) {
} else {
ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
}
+
+ // After stopping the HMS:
+ // 1. We should still be able to initialize the catalog.
+ // 2. Attempts to fetch the DB UUID will fail.
+ ASSERT_OK(hms_->Stop());
+ hms_catalog_.reset(new HmsCatalog(kMasterAddrs));
+ ASSERT_OK(hms_catalog_->Start());
+ ASSERT_TRUE(hms_catalog_->GetUuid(nullptr).IsNotSupported());
+
+ // But if we start the HMS back up, we should be able to eventually get the
+ // DB UUID, though we need to account for the possibility that this HMS may
+ // not support it.
+ ASSERT_OK(hms_->Start());
+ string uuid2;
+ ASSERT_EVENTUALLY([&] {
+ Status s2 = hms_catalog_->GetUuid(&uuid2);
+ if (s.ok()) {
+ ASSERT_OK(s2);
+ } else {
+ ASSERT_TRUE(s2.IsNotSupported());
+ }
+ });
+ ASSERT_EQ(uuid, uuid2);
}
} // namespace hms
diff --git a/src/kudu/hms/hms_catalog.cc b/src/kudu/hms/hms_catalog.cc
index 0fb46a7..686bcdd 100644
--- a/src/kudu/hms/hms_catalog.cc
+++ b/src/kudu/hms/hms_catalog.cc
@@ -20,6 +20,7 @@
#include <iostream>
#include <iterator>
#include <map>
+#include <mutex>
#include <string>
#include <type_traits>
#include <utility>
@@ -45,8 +46,8 @@
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/slice.h"
+#include "kudu/util/thread.h"
-using boost::none;
using boost::optional;
using std::move;
using std::string;
@@ -103,7 +104,8 @@ namespace kudu {
namespace hms {
HmsCatalog::HmsCatalog(string master_addresses)
- : master_addresses_(std::move(master_addresses)) {
+ : master_addresses_(std::move(master_addresses)),
+ running_(1) {
}
HmsCatalog::~HmsCatalog() {
@@ -111,6 +113,7 @@ HmsCatalog::~HmsCatalog() {
}
Status HmsCatalog::Start(HmsClientVerifyKuduSyncConfig verify_service_config) {
+ running_.Reset(1);
vector<HostPort> addresses;
RETURN_NOT_OK(ParseUris(FLAGS_hive_metastore_uris, &addresses));
@@ -126,25 +129,18 @@ Status HmsCatalog::Start(HmsClientVerifyKuduSyncConfig verify_service_config) {
RETURN_NOT_OK(ha_client_.Start(std::move(addresses), std::move(options)));
- // Fetch the UUID of the HMS DB, if available.
- string uuid;
- Status uuid_status = ha_client_.Execute([&] (HmsClient* client) {
- return client->GetUuid(&uuid);
- });
- if (uuid_status.ok()) {
- VLOG(1) << "Connected to HMS with uuid " << uuid;
- uuid_ = std::move(uuid);
- } else if (uuid_status.IsNotSupported()) {
- VLOG(1) << "Unable to fetch UUID for HMS: " << uuid_status.ToString();
- } else {
- // If we couldn't connect to the HMS at all, fail.
- return uuid_status;
- }
+ RETURN_NOT_OK(Thread::Create("hms_catalog", "fetch_uuid",
+ &HmsCatalog::LoopInitializeUuid, this,
+ &uuid_initializing_thread_));
return Status::OK();
}
void HmsCatalog::Stop() {
+ running_.CountDown();
ha_client_.Stop();
+ if (uuid_initializing_thread_) {
+ uuid_initializing_thread_->Join();
+ }
}
Status HmsCatalog::CreateTable(const string& id,
@@ -305,6 +301,7 @@ Status HmsCatalog::GetNotificationEvents(int64_t last_event_id, int max_events,
}
Status HmsCatalog::GetUuid(string* uuid) {
+ std::lock_guard<simple_spinlock> l(uuid_lock_);
if (!uuid_) {
return Status::NotSupported("No HMS UUID available");
}
@@ -451,6 +448,26 @@ bool HmsCatalog::ValidateUris(const char* flag_name, const string& metastore_uri
return s.ok();
}
+void HmsCatalog::LoopInitializeUuid() {
+ do {
+ // Fetch the UUID of the HMS DB, if available.
+ string uuid;
+ Status s = ha_client_.Execute([&] (HmsClient* client) {
+ return client->GetUuid(&uuid);
+ });
+ if (s.ok()) {
+ VLOG(1) << "Connected to HMS with uuid " << uuid;
+ std::lock_guard<simple_spinlock> l(uuid_lock_);
+ uuid_ = std::move(uuid);
+ } else if (s.IsNotSupported()) {
+ VLOG(1) << "Unable to fetch UUID for HMS: " << s.ToString();
+ return;
+ }
+ // If we couldn't connect to the HMS at all, sleep and try again.
+ VLOG(1) << "Couldn't connect to HMS: " << s.ToString();
+ } while (!running_.WaitFor(MonoDelta::FromSeconds(1)));
+}
+
bool HmsCatalog::IsEnabled() {
return !FLAGS_hive_metastore_uris.empty();
}
diff --git a/src/kudu/hms/hms_catalog.h b/src/kudu/hms/hms_catalog.h
index 981da96..95c44c1 100644
--- a/src/kudu/hms/hms_catalog.h
+++ b/src/kudu/hms/hms_catalog.h
@@ -25,15 +25,19 @@
#include <gtest/gtest_prod.h>
#include "kudu/gutil/port.h"
+#include "kudu/gutil/ref_counted.h"
#include "kudu/hms/hive_metastore_types.h"
#include "kudu/hms/hms_client.h"
#include "kudu/thrift/client.h"
+#include "kudu/util/countdown_latch.h"
+#include "kudu/util/locks.h"
#include "kudu/util/status.h"
namespace kudu {
class HostPort;
class Schema;
+class Thread;
namespace hms {
@@ -137,7 +141,9 @@ class HmsCatalog {
// HMS itself changes hostnames, etc.
//
// NOTE: this is not implemented in Hive 2.x but may be backported into some
- // vendor releases (eg CDH6). This function may return Status::NotSupported().
+ // vendor releases (eg CDH6). This function may return Status::NotSupported()
+ // in such cases, or if the HmsCatalog has been unable to connect to the HMS
+ // (i.e. the HMS has been down for the entirety of the HmsCatalog's lifespan).
Status GetUuid(std::string* uuid) WARN_UNUSED_RESULT;
// Validates the hive_metastore_uris gflag.
@@ -186,6 +192,14 @@ class HmsCatalog {
Status DropTable(const std::string& name,
const hive::EnvironmentContext& env_ctx) WARN_UNUSED_RESULT;
+ // Repeatedly tries to fetch the HMS' DB UUID, sleeping in between attempts.
+ //
+ // The loop terminates when one of the following conditions are met:
+ // 1. The UUID is fetched successfully.
+ // 2. running_ counts down to 0.
+ // 3. The HMS responds with NotSupported.
+ void LoopInitializeUuid();
+
// Parses a Hive Metastore URI string into a sequence of HostPorts.
static Status ParseUris(const std::string& metastore_uris, std::vector<HostPort>* hostports);
@@ -195,8 +209,20 @@ class HmsCatalog {
// Kudu master addresses.
const std::string master_addresses_;
+ // HmsClient instance used for remote HMS requests.
thrift::HaClient<hms::HmsClient> ha_client_;
+ // Background thread that tries to initialize 'uuid_'.
+ scoped_refptr<Thread> uuid_initializing_thread_;
+
+ // Set to 0 when the HmsCatalog instance has been stopped.
+ CountDownLatch running_;
+
+ // Protects 'uuid_'.
+ simple_spinlock uuid_lock_;
+
+ // The UUID of the remote HMS instance. Will be boost::none if the HMS is
+ // down, or if the HMS does not support UUIDs.
boost::optional<std::string> uuid_;
};
[kudu] 01/02: docs: Add simplest possible Spark SQL example
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit d25c6f11ca8611bbe30b6951fbda7397b0a3ac88
Author: Mike Percy <mp...@apache.org>
AuthorDate: Tue Jul 31 15:53:52 2018 -0700
docs: Add simplest possible Spark SQL example
Often I look for a simple "hello world" example in the Kudu Spark docs
and I remember that there isn't one. I've added a quick-and-dirty Spark
SQL example.
Change-Id: I2cf4c00f3a1dc92fd93458aa3c1b1d2cd4f38f78
Reviewed-on: http://gerrit.cloudera.org:8080/11095
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
docs/developing.adoc | 37 ++++++++++++++++++++++++++++---------
1 file changed, 28 insertions(+), 9 deletions(-)
diff --git a/docs/developing.adoc b/docs/developing.adoc
index 89db3e8..210db59 100644
--- a/docs/developing.adoc
+++ b/docs/developing.adoc
@@ -109,7 +109,26 @@ on the link:http://kudu.apache.org/releases/[releases page].
spark-shell --packages org.apache.kudu:kudu-spark2_2.11:1.10.0
----
-then import kudu-spark and create a dataframe:
+Below is a minimal Spark SQL "select" example for a Kudu table created with
+Impala in the "default" database. We first import the kudu spark package,
+then create a DataFrame, and then create a view from the DataFrame. After those
+steps, the table is accessible from Spark SQL.
+
+[source,scala]
+----
+import org.apache.kudu.spark.kudu._
+
+// Create a DataFrame that points to the Kudu table we want to query.
+val df = spark.read.options(Map("kudu.master" -> "master1.foo.com,master2.foo.com,master3.foo.com",
+ "kudu.table" -> "default.my_table")).kudu
+// Create a view from the DataFrame to make it accessible from Spark SQL.
+df.createOrReplaceTempView("my_table")
+// Now we can run Spark SQL queries against our view of the Kudu table.
+spark.sql("select * from my_table").show()
+----
+
+Below is a more sophisticated example that includes both reads and writes:
+
[source,scala]
----
import org.apache.kudu.client._
@@ -124,14 +143,14 @@ val df = spark.read
df.select("id").filter("id >= 5").show()
// ...or register a temporary table and use SQL
-df.registerTempTable("kudu_table")
+df.createOrReplaceTempView("kudu_table")
val filteredDF = spark.sql("select id from kudu_table where id >= 5").show()
// Use KuduContext to create, delete, or write to Kudu tables
val kuduContext = new KuduContext("kudu.master:7051", spark.sparkContext)
-// Create a new Kudu table from a dataframe schema
-// NB: No rows from the dataframe are inserted into the table
+// Create a new Kudu table from a DataFrame schema
+// NB: No rows from the DataFrame are inserted into the table
kuduContext.createTable(
"test_table", df.schema, Seq("key"),
new CreateTableOptions()
@@ -170,15 +189,15 @@ kuduContext.deleteTable("unwanted_table")
=== Upsert option in Kudu Spark
The upsert operation in kudu-spark supports an extra write option of `ignoreNull`. If set to true,
-it will avoid setting existing column values in Kudu table to Null if the corresponding dataframe
+it will avoid setting existing column values in Kudu table to Null if the corresponding DataFrame
column values are Null. If unspecified, `ignoreNull` is false by default.
[source,scala]
----
-val dataDF = spark.read
+val dataFrame = spark.read
.options(Map("kudu.master" -> "kudu.master:7051", "kudu.table" -> simpleTableName))
.format("kudu").load
-dataDF.registerTempTable(simpleTableName)
-dataDF.show()
+dataFrame.createOrReplaceTempView(simpleTableName)
+dataFrame.show()
// Below is the original data in the table 'simpleTableName'
+---+---+
|key|val|
@@ -191,7 +210,7 @@ val nullDF = spark.createDataFrame(Seq((0, null.asInstanceOf[String]))).toDF("ke
val wo = new KuduWriteOptions
wo.ignoreNull = true
kuduContext.upsertRows(nullDF, simpleTableName, wo)
-dataDF.show()
+dataFrame.show()
// The val field stays unchanged
+---+---+
|key|val|