What is big data - Infographics 07-19

What is big data - Infographics 

View at the original source

How-to: Tune Your Apache Spark Jobs (Part 2) 04-28

How-to: Tune Your Apache Spark Jobs (Part 2)

In the conclusion to this series, learn how resource tuning, parallelism, and data representation affect Spark job performance.

In this post, we’ll finish what we started in “How to Tune Your Apache Spark Jobs (Part 1)”. I’ll try to cover pretty much everything you could care to know about making a Spark program run fast. In particular, you’ll learn about resource tuning, or configuring Spark to take advantage of everything the cluster has to offer. Then we’ll move to tuning parallelism, the most difficult as well as most important parameter in job performance. Finally, you’ll learn about representing the data itself, in the on-disk form which Spark will read (spoiler alert: use Apache Avro or Apache Parquet) as well as the in-memory format it takes as it’s cached or moves through the system.

Tuning Resource Allocation

The Spark user list is a litany of questions to the effect of “I have a 500-node cluster, but when I run my application, I see only two tasks executing at a time. HALP.” Given the number of parameters that control Spark’s resource utilization, these questions aren’t unfair, but in this section you’ll learn how to squeeze every last bit of juice out of your cluster. The recommendations and configurations here differ a little bit between Spark’s cluster managers (YARN, Mesos, and Spark Standalone), but we’re going to focus only on YARN, which Cloudera recommends to all users.

The two main resources that Spark (and YARN) think about are CPU and memory. Disk and network I/O, of course, play a part in Spark performance as well, but neither Spark nor YARN currently do anything to actively manage them.

Every Spark executor in an application has the same fixed number of cores and same fixed heap size. The number of cores can be specified with the --executor-cores flag when invoking spark-submit, spark-shell, and pyspark from the command line, or by setting the spark.executor.cores property in the spark-defaults.conf file or on aSparkConf object. Similarly, the heap size can be controlled with the --executor-cores flag or thespark.executor.memory property. The cores property controls the number of concurrent tasks an executor can run. --executor-cores 5 means that each executor can run a maximum of five tasks at the same time. The memory property impacts the amount of data Spark can cache, as well as the maximum sizes of the shuffle data structures used for grouping, aggregations, and joins.

The --num-executors command-line flag or spark.executor.instances configuration property control the number of executors requested. Starting in CDH 5.4/Spark 1.3, you will be able to avoid setting this property by turning ondynamic allocation with the spark.dynamicAllocation.enabled property. Dynamic allocation enables a Spark application to request executors when there is a backlog of pending tasks and free up executors when idle.

It’s also important to think about how the resources requested by Spark will fit into what YARN has available. The relevant YARN properties are:

• yarn.nodemanager.resource.memory-mb controls the maximum sum of memory used by the containers on each node.
• yarn.nodemanager.resource.cpu-vcores controls the maximum sum of cores used by the containers on each node.

Asking for five executor cores will result in a request to YARN for five virtual cores. The memory requested from YARN is a little more complex for a couple reasons: 

• --executor-memory/spark.executor.memory controls the executor heap size, but JVMs can also use some memory off heap, for example for interned Strings and direct byte buffers. The value of thespark.yarn.executor.memoryOverhead property is added to the executor memory to determine the full memory request to YARN for each executor. It defaults to max(384, .07 * spark.executor.memory).
• YARN may round the requested memory up a little. YARN’s yarn.scheduler.minimum-allocation-mb andyarn.scheduler.increment-allocation-mb properties control the minimum and increment request values respectively.

The following (not to scale with defaults) shows the hierarchy of memory properties in Spark and YARN:

And if that weren’t enough to think about, a few final concerns when sizing Spark executors:

• The application master, which is a non-executor container with the special capability of requesting containers from YARN, takes up resources of its own that must be budgeted in. In yarn-client mode, it defaults to a 1024MB and one vcore. In yarn-cluster mode, the application master runs the driver, so it’s often useful to bolster its resources with the --driver-memory and --driver-cores properties.
• Running executors with too much memory often results in excessive garbage collection delays. 64GB is a rough guess at a good upper limit for a single executor.
• I’ve noticed that the HDFS client has trouble with tons of concurrent threads. A rough guess is that at most five tasks per executor can achieve full write throughput, so it’s good to keep the number of cores per executor below that number.
• Running tiny executors (with a single core and just enough memory needed to run a single task, for example) throws away the benefits that come from running multiple tasks in a single JVM. For example, broadcast variables need to be replicated once on each executor, so many small executors will result in many more copies of the data.

To hopefully make all of this a little more concrete, here’s a worked example of configuring a Spark app to use as much of the cluster as possible: Imagine a cluster with six nodes running NodeManagers, each equipped with 16 cores and 64GB of memory. The NodeManager capacities, yarn.nodemanager.resource.memory-mb andyarn.nodemanager.resource.cpu-vcores, should probably be set to 63 * 1024 = 64512 (megabytes) and 15 respectively. We avoid allocating 100% of the resources to YARN containers because the node needs some resources to run the OS and Hadoop daemons. In this case, we leave a gigabyte and a core for these system processes. Cloudera Manager helps by accounting for these and configuring these YARN properties automatically.

The likely first impulse would be to use --num-executors 6 --executor-cores 15 --executor-memory 63G. However, this is the wrong approach because:

• 63GB + the executor memory overhead won’t fit within the 63GB capacity of the NodeManagers.
• The application master will take up a core on one of the nodes, meaning that there won’t be room for a 15-core executor on that node.
• 15 cores per executor can lead to bad HDFS I/O throughput.

A better option would be to use --num-executors 17 --executor-cores 5 --executor-memory 19G. Why?

• This config results in three executors on all nodes except for the one with the AM, which will have two executors.
• --executor-memory was derived as (63/3 executors per node) = 21.  21 * 0.07 = 1.47.  21 – 1.47 ~ 19.

Tuning Parallelism

Spark, as you have likely figured out by this point, is a parallel processing engine. What is maybe less obvious is that Spark is not a “magic” parallel processing engine, and is limited in its ability to figure out the optimal amount of parallelism. Every Spark stage has a number of tasks, each of which processes data sequentially. In tuning Spark jobs, this number is probably the single most important parameter in determining performance.

How is this number determined? The way Spark groups RDDs into stages is described in the previous post. (As a quick reminder, transformations like repartition and reduceByKey induce stage boundaries.) The number of tasks in a stage is the same as the number of partitions in the last RDD in the stage. The number of partitions in an RDD is the same as the number of partitions in the RDD on which it depends, with a couple exceptions: thecoalescetransformation allows creating an RDD with fewer partitions than its parent RDD, the union transformation creates an RDD with the sum of its parents’ number of partitions, and cartesian creates an RDD with their product.

What about RDDs with no parents? RDDs produced by textFile or hadoopFile have their partitions determined by the underlying MapReduce InputFormat that’s used. Typically there will be a partition for each HDFS block being read. Partitions for RDDs produced by parallelize come from the parameter given by the user, orspark.default.parallelism if none is given.

To determine the number of partitions in an RDD, you can always call rdd.partitions().size().

The primary concern is that the number of tasks will be too small. If there are fewer tasks than slots available to run them in, the stage won’t be taking advantage of all the CPU available. 

A small number of tasks also mean that more memory pressure is placed on any aggregation operations that occur in each task. Any join, cogroup, or *ByKey operation involves holding objects in hashmaps or in-memory buffers to group or sort. join, cogroup, and groupByKey use these data structures in the tasks for the stages that are on the fetching side of the shuffles they trigger. reduceByKey and aggregateByKey use data structures in the tasks for the stages on both sides of the shuffles they trigger.

When the records destined for these aggregation operations do not easily fit in memory, some mayhem can ensue. First, holding many records in these data structures puts pressure on garbage collection, which can lead to pauses down the line. Second, when the records do not fit in memory, Spark will spill them to disk, which causes disk I/O and sorting. This overhead during large shuffles is probably the number one cause of job stalls I have seen at Cloudera customers.

So how do you increase the number of partitions? If the stage in question is reading from Hadoop, your options are:

• Use the repartition transformation, which will trigger a shuffle.
• Configure your InputFormat to create more splits.
• Write the input data out to HDFS with a smaller block size.

If the stage is getting its input from another stage, the transformation that triggered the stage boundary will accept anumPartitions argument, such as

1	val rdd2 = rdd1.reduceByKey(_ + _, numPartitions = X)

What should “X” be? The most straightforward way to tune the number of partitions is experimentation: Look at the number of partitions in the parent RDD and then keep multiplying that by 1.5 until performance stops improving. 

There is also a more principled way of calculating X, but it’s difficult to apply a priori because some of the quantities are difficult to calculate. I’m including it here not because it’s recommended for daily use, but because it helps with understanding what’s going on. The main goal is to run enough tasks so that the data destined for each task fits in the memory available to that task.

The memory available to each task is (spark.executor.memory * spark.shuffle.memoryFraction *spark.shuffle.safetyFraction)/spark.executor.cores. Memory fraction and safety fraction default to 0.2 and 0.8 respectively.

The in-memory size of the total shuffle data is harder to determine. The closest heuristic is to find the ratio between Shuffle Spill (Memory) metric and the Shuffle Spill (Disk) for a stage that ran. Then multiply the total shuffle write by this number. However, this can be somewhat compounded if the stage is doing a reduction:

Then round up a bit because too many partitions is usually better than too few partitions.

In fact, when in doubt, it’s almost always better to err on the side of a larger number of tasks (and thus partitions). This advice is in contrast to recommendations for MapReduce, which requires you to be more conservative with the number of tasks. The difference stems from the fact that MapReduce has a high startup overhead for tasks, while Spark does not.

Slimming Down Your Data Structures

Data flows through Spark in the form of records. A record has two representations: a deserialized Java object representation and a serialized binary representation. In general, Spark uses the deserialized representation for records in memory and the serialized representation for records stored on disk or being transferred over the network. There is work planned to store some in-memory shuffle data in serialized form.

The spark.serializer property controls the serializer that’s used to convert between these two representations. The Kryo serializer, org.apache.spark.serializer.KryoSerializer, is the preferred option. It is unfortunately not the default, because of some instabilities in Kryo during earlier versions of Spark and a desire not to break compatibility, but the Kryo serializer should always be used

The footprint of your records in these two representations has a massive impact on Spark performance. It’s worthwhile to review the data types that get passed around and look for places to trim some fat.

Bloated deserialized objects will result in Spark spilling data to disk more often and reduce the number of deserialized records Spark can cache (e.g. at the MEMORY storage level). The Spark tuning guide has a great section on slimming these down.

Bloated serialized objects will result in greater disk and network I/O, as well as reduce the number of serialized records Spark can cache (e.g. at the MEMORY_SER storage level.)  The main action item here is to make sure to register any custom classes you define and pass around using the SparkConf#registerKryoClasses API.

Data Formats

Whenever you have the power to make the decision about how data is stored on disk, use an extensible binary format like Avro, Parquet, Thrift, or Protobuf. Pick one of these formats and stick to it. To be clear, when one talks about using Avro, Thrift, or Protobuf on Hadoop, they mean that each record is a Avro/Thrift/Protobuf struct stored in asequence file. JSON is just not worth it. 

Every time you consider storing lots of data in JSON, think about the conflicts that will be started in the Middle East, the beautiful rivers that will be dammed in Canada, or the radioactive fallout from the nuclear plants that will be built in the American heartland to power the CPU cycles spent parsing your files over and over and over again. Also, try to learn people skills so that you can convince your peers and superiors to do this, too.

View at the original source

What is a data lake? 04-27

What is a data lake?

You’ve probably heard of data warehousing, but now there’s a newer phrase doing the rounds, and it’s one you’re likely to hear more in the future if you’re involved in big data: ‘Data Lakes’.

So what are they? Well, the best way to describe them is to compare them to data warehouses, because the difference is very much the same as between storing something in a warehouse and storing something in a lake.

In a warehouse, everything is archived and ordered in a defined way – the products are inside containers, the containers on shelves, the shelves are in rows, and so on. This is the way that data is stored in a traditional data warehouse.

In a data lake, everything is just poured in, in an unstructured way. A molecule of water in the lake is equal to any other molecule and can be moved to any part of the lake where it will feel equally at home.

This means that data in a lake has a great deal of agility – another word which is becoming more frequently used these days – in that it can be configured or reconfigured as necessary, depending on the job you want to do with it.

A data lake contains data in its rawest form – fresh from capture, and unadulterated by processing or analysis.

It uses what is known as object-based storage, because each individual piece of data is treated as an object, made up of the information itself packaged together with its associated metadata, and a unique identifier.

No piece of information is “higher-level” than any other, because it is not a hierarchically archived system, like a warehouse – it is basically a big free-for-all, as water molecules exist in a lake.

The term is thought to have first been used by Pentaho CTO James Dixon in 2011, who didn’t invent the concept but gave a name to the type of innovative data architecture solutions being put to use by companies such as Google and Facebook.

It didn’t take long for the name to make it into marketing material. Pivotal refer to their product as a “business data lake” and Hortonworks include it in the name of their service, Hortonworks Datalakes.

It is a practice which is expected to become more popular in the future, as more organizations become aware of the increased agility afforded by storing data in data lakes rather than strict hierarchical databases.

For example, the way that data is stored in a database (its “schema”) is often defined in the early days of the design of a data strategy.  The needs and priorities of the organization may well change as time goes on.

One way of thinking about it is that data stored without structure can be more quickly shaped into whatever form it is needed, than if you first have to disassemble the previous structure before reassembling it.

Another advantage is that the data is available to anyone in the organization, and can be analyzed and interrogated via different tools and interfaces as appropriate for each job.

It also means that all of an organization’s data is kept in one place – rather than having separate data stores for individual departments or applications, as is often the case.

This brings its own advantages and disadvantages – on the one hand, it makes auditing and compliancy simpler, with only one store to manage. On the other, there are obvious security implications if you’re keeping “all your eggs in one basket”.

Data lakes are usually built within the Hadoop framework, as the datasets they are comprised of are “big” and need the volume of storage offered by distributed systems.

A lot of it is theoretical at the moment because there are very few organizations which are ready to make the move to keeping all of their data in a lake. Many are bogged down in a “data swamp” – hard-to-navigate mishmashes of land and water where their data has been stored in various, uncoordinated ways over the years.

And it has its critics of course – some say that the name itself is a problem (and I am inclined to agree) as it implies a lack of architectural awareness, when a more careful consideration of data architecture is what’s really needed when designing new solutions.

But for better or worse, it is a term that you will probably be hearing more of in the near future if you’re involved in big data and business intelligence.

Are you ready to dive head first into the data lake or do you prefer to keep your data high and dry?

View at the original source

Big data redefines the traditional scientific methods used in medicine 02-15

Big data redefines the traditional scientific methods used in medicine

Healthcare professionals are applying big data and analytics to clinical challenges. This is just the beginning of a redefinition in the traditional scientific methods used in medicine. 

IBM Watson

 Image: Bob Goldberg/courtesy of IBM

Stanford University will host a big data in biomedicine conference May 20-22, 2015 for medical researchers hailing from colleges and universities, hospitals, government, and industry. The goals are to encourage collaboration, address challenges, and identify actionable steps for harnessing big data in healthcare.

There are plenty of incentives. Whether through mega-scientific computing projects that process petabytes of data or through more informal ways of looking at data and analyzing it in new ways to reach outcomes that were previously unattainable, medicine is marching forward in applying big data and analytics to clinical challenges.

For instance, at Lucile Packard Children's Hospital Stanford in 2011, a young girl from Reno, Nevada, was flown by helicopter to the hospital, where she was admitted to the intensive care unit. The girl had lupus, which attacks the body's healthy tissues and can cause permanent kidney damage. An interdisciplinary team of doctors had to weigh the risks of using a coagulant that could thin blood and help prevent clots against the counter risks of complicating surgery, causing a stroke or creating a bleed into an organ. The team needed data.

A young physician named Jennifer Frankovich resorted to using a database of children with lupus that she had been helping to build. Part of the database work had entailed digitalizing charts and making data searchable with keywords. Through database searches, Dr. Frankovich was able to look at every pediatric lupus patient who had come through the hospital to see how many of them developed blood clots, and what the risk factors were. From there, she could calculate whether the risks of a blood clot in her current patient justified the risks of prescribing an anti-coagulant. The calculations indicated that the risk was worth taking, and the patient was given an anti-coagulant. The patient immediately showed signs of improvement.

Atul Butte, an entrepreneur and associate professor of pediatrics at the Stanford School of Medicine, compared Dr. Frankovich's work to a "seismic shift" happening in medicine. "The idea here is, the scientific method itself is growing obsolete," Butte said.

This scientific method as it has existed for decades and continues to exist in medicine today consists of a team of eminently qualified specialists from a variety of medical fields consulting with each other and sharing their collective experiences of treatment options and outcomes for the patient. In cases where unusual circumstances or risks present themselves, medical literature and empirical evidence that the scientific method demands are often missing. This happened in the Stanford lupus case, and this is where Dr. Frankovich was able to fill in the blanks with insights from data.

Is this the end of the story? Not quite.

Administrators at that hospital still feel it is safer to trust the wisdom of a team of doctors in urgent cases than to search medical records for data about what's worked in the past. In a January 2015 interview with NPR, Dr. Frankovich agreed, noting that "analyzing data is complicated and requires specific expertise. What if the search engine has bugs, or the records are transcribed incorrectly? There's just too much room for error....It's going to take a system to interpret the data, and that's what we don't have yet."

This comes full circle to the upcoming big data conference at Stanford this spring. The conference announcement states, "While other industries have been far more successful at harnessing the value from large-scale integration and analysis of big data, health care is just getting its feet wet. Yes, providers and payers are increasingly investing in their analytical capabilities to help them make better sense of the changing health care environment, but it is still early days."

True enough, but in healthcare settings like the Cleveland Clinic, doctors and medical practitioners are already making use of big data and analytics that diagnose conditions and prescribe treatments. What the analytics say is simply entered into the discussions that interdisciplinary teams of doctors have when they review patients. And while data quality and integration issues in healthcare continue to persist, there is an unmistakable beginning of a redefinition in the traditional scientific methods used in medicine.

View at the original source

Applications Drive The Biggest Money In Big Data 01-09

Applications Drive The Biggest Money In Big Data

The real money in Big Data has nothing to do with selling Hadoop.

We're still fixating on all the wrong Big Data startups. Hortonworks, one of the primary companies behind Hadoop, recently went public to great fanfare and a $1.2 billion valuation. But Hortonworks and the rest of the so-called Big Data startups are actually some of the least interesting Big Data companies.

In fact, of the current crop of 40 startups valued at more than $1 billion, virtually none of them sell Big Data technology like Hadoop. But all of them make heavy use of data - lots of it - to deliver a wide array of services.

As consultant Peter Goldmacher declared back in 2013, the biggest winners in Big Data are the "business people that have identified opportunities to use data to create new opportunities or disrupt legacy business models." As we enter 2015, expect to see data double the number of billion-dollar startups even as public companies learn to grow through data, as well.

Do-It-Yourself Software Loses Its Luster

It used to be enough for a vendor to ship software and abandon the customer to figure it out (or pay hefty sums of money in consulting fees). SAP, for example, has made billions in revenue by shipping complex software and having customers shell out multiples of the software license fee for high-priced consultants to make sense of its Byzantine software.

That sort of strategy doesn't work very well anymore.

Forget startups for a moment. If we look at the stock prices of various data-related companies, investors are paying a premium for companies like Tableau and Qlik that make data easy to consume:

The companies rising the most include Tableau, Qlik and MicroStrategy, which provide tools to visualize data, while companies that tend to sell infrastructure like IBM and Teradata largely skidded through the year. (In fact, IBM is on its second year as one of the Dow Jones worst performers.)

Data Begets Billions

The analysis isn't perfect, of course. For example, though IBM sells a lot of core infrastructure it also has a Business Intelligence business. Oracle, for its part, plays in many camps, with a strong applications business to make up for its stalling database business.

But where the shift to Big Data really becomes apparent is in the Wall Street Journal's burgeoning billion-dollar startup club. As the Journal's Christopher Mims points out, "2014 was the year tech startup valuations went on a tear without precedent." 

It was also the year that tech startups put data to use at unprecedented levels.

No, not in the old-school Big Data way. When you review most lists of the "top 10 Big Data companies" they focus on those that sell Big Data technology. Among the top-15 most valuable startups, only Cloudera (and maybe Palantir) counts as a Big Data startup in this old sense of the word. 

Source: Wall Street Journal

Comb through the rest of the top-40 most valuable startups and you add MongoDB and Good Data. At face value this seems to suggest that Big Data really isn't that big of a deal.

Back to Goldmacher.

In Goldmacher's world, the "Big winners" in Big Data are "infrastructure providers like the Hadoop vendors and the NoSQL vendors," the "Bigger winners" are "the Apps and Analytics vendors that abstract the complexity of working with very complicated underlying technologies into a user friendly front end."

There's An App For That

But the "Biggest winners," as noted above, are companies like Uber, Stripe and Airbnb that have figured out how to "leverage data as an asset," thereby up-ending old industries and setting themselves apart. Look through the list of the top-40 most valuable startups and nearly all of them have this in common: they understand and leverage Big Data.

As we enter 2015, data will become more important than ever. It won't, however, be easy to track, because there's no meaningful "Big Data" category of vendors. Instead, data will transform industries as diverse as retail and healthcare, crowning multitudes of billion-dollar startups and billion-dollar revenue streams along the way.

View at the original source