Data Generation

The first stage in big data is the generation of data. The Internet is full of data. We can see it everywhere: searches, posts on different forums, registers of conversations, and much more. By analyzing data available on the Internet, we can learn much. Taken individually, we can discover the habits of a user, but taken as a whole, we can discover different trends, habits, or even emotional facts through sentiment analysis. The data is complex and diverse because it is obtained from different locations around the world and the sources are distributed. At this moment, the main origins of big data are operations and commercial data from companies, logistic and sensors data from the Internet of Things (IoT), social platforms, and data from the research fields, among others. This amount of data cannot be handled by traditional IT architectures and infrastructures. Moreover, it cannot be analyzed using traditional systems.

Data Collection

After data is generated, the next step is to collect it from different sources. The following are data collection approaches.

• Log files. The system that represents the origin of data generates automatic log files at a certain time. Usually, the log files contain information about activities on that system. For example, web servers usually have three types of log files that contain information about users’ behaviors: public format (NCSA), expanded format (W3C), and IIS format (Microsoft)—all in ASCII format. Sometimes, log information is stored in databases because the queries are more efficient. Also, log information could be financial data, or traffic from a network.

• Detection. Sensors are used to detect a certain type of data and to transmit it through a channel to a collecting site.

• Techniques for collecting data. Includes web crawlers (used in search systems when web pages are downloaded and stored), systems for word segmentation, tasks, and index systems.

We will not continue to talk about how data is transmitted to storage systems. Next, we discuss some of the tools for big data storage.

Data Storage

In big data, the main approach of data storage is distributed storage systems, which make use of more servers for data storage and copies of data. Commonly, data is fragmented into lower pieces that are stored on different servers. The storage systems need to have at least the following characteristics.

• Consistency. This ensures that all copies of the same original data are the same.

• Availability. This ensures that data is available anytime, even in the event of a server or virtual machine failure. This is the reason why data is fragmented and copies of the same data are stored at different locations.

• Fault tolerance. This ensures that the data is available and can be managed even if there is a network failure.

Big data storage systems have three levels on the top of the architecture.

• File systems represent the way in which files on a storage systems are named, the logical place of their storage, how they could be retrieved, and so on. Examples of file systems include GFS (Google File System) and Colossus from Google. HDFS (Hadoop Data File System) and Kosmosfs are open source and derived from GFS, Cosmos from Microsoft, Haystack from Facebook, and so forth.

• Traditional databases are collections of data organized (generally they are stored into tables) to be easily manipulated. In the last few years, however, NoSQL (Not Only SQL) databases, which are not relational, are being used more and more. Their targets are large sets of distributed data. You will see some examples of NoSQL databases in the next subsection.

• Programming models represent sets of concepts used for creating software applications. MapReduce is a well-known example of this programming model.

Polymorphic Implementation

Previously, we changed the type of distrMapReduce, because it was necessary that the type of the MapReduce skeleton be matched in the word-counting example. The type could be changed without changing the implementation. The following is the polymorphic version of distrMapreduce.

distrMapReduce :: (Serializable k1, Serializable v1, Serializable k2,
                   Serializable v2, Serializable v3, Ord k2)
               => Closure (MapReduce k1 v1 k2 v2 v3)
               -> [NodeId]
               -> ((Map k1 v1 -> Process (Map k2 v3)) -> Process a)
               -> Process a

Anyway, there is a little problem with the generalization. Let’s think about mappers in the general form. What do they need to do? Well, at first sight, it should expect a message that has a specific type. For type matching, it should know something about the type (k1, v1). The next step is to send a list whose type is [(k2, v2)] to the master when a message comes. This is possible only when the map knows the way it could serialize the values with type [(k2, v2)].

distrMapReduce obtains this information from the Serializable type class’s constraints. Unfortunately, in Haskell there is no explicit way such that the arguments to be handled, even more providing a way for serializing them for being shipped to the mapper nodes. We can define a type class constraint as an explicit dictionary, however.

data SerializableDict a where
    SerializableDict :: Serializable a => SerializableDict a

The objects whose type is SerializableDict cannot be directly serialized, but static SerializableDicts can. So, definition of serializing become:

distrMapReduce :: forall k1 k2 v1 v2 v3 a.
                  (Serializable k1, Serializable v1, Serializable k2,
                   Serializable v2, Serializable v3, Ord k2)
               => Static (SerializableDict (k1, v1))
               -> Static (SerializableDict [(k2, v2)])
               -> Closure (MapReduce k1 v1 k2 v2 v3)
               -> [NodeId]
               -> ((Map k1 v1 -> Process (Map k2 v3)) -> Process a)
               -> Process a

Maybe it is a little complicated, but the change is requiring static type information to ship this type of information to the mappers. We omitted the implementation. It could be found in distributed-process-demos package; the basic principles are explained in the documentation of the distributed-static package.

The polymorphic version of distrMapReduce has the same difficulty as monomorphic version ; for example, “distributed word counting” can be implemented as follows.

dictIn :: SerializableDict (FilePath, Document)
dictIn = SerializableDict

dictOut :: SerializableDict [(Word, Frequency)]
dictOut = SerializableDict

countWords_ :: () -> MapReduce FilePath Document Word Frequency Frequency
countWords_ () = countWords

remotable ['dictIn, 'dictOut, 'countWords_]

distrCountWords :: [NodeId] -> Map FilePath Document -> Process (Map Word Frequency)
distrCountWords mappers input =
  distrMapReduce $(mkStatic 'dictIn)
                 $(mkStatic 'dictOut)
                 ($(mkClosure 'countWords_) ())
                 mappers
                 (iteration -> iteration input)

Creating the necessary SerializableDicts is easy (there is only one constructor for SerializableDict, and it doesn’t take any arguments!). Note that the word counter only calls the iteration function once; this will not be true for distributed k-means.

Distributed k-means

The following presents the distributed version of k-means . There are not very many modifications to the initial (local) implementation: go is added, and the rest (segments, dividePoints, pointsPerMapper, and numPoints) show where every segment needs to go to the corresponding node from map.

dictIn :: SerializableDict ((Int, Int), [Cluster])
dictIn = SerializableDict

dictOut :: SerializableDict [(Cluster, Point)]
dictOut = SerializableDict

remotable ['kmeans, 'dictIn, 'dictOut]

distrKMeans :: Array Int Point
            -> [Cluster]
            -> [NodeId]
            -> Int
            -> Process (Map Cluster ([Point], Point))
distrKMeans points cs mappers iterations =
    distrMapReduce $(mkStatic 'dictIn)
                   $(mkStatic 'dictOut)
                   ($(mkClosure 'kmeans) points)
                   mappers
                   (go (iterations - 1))
  where
    go :: Int
       -> (Map (Int, Int) [Cluster] -> Process                                                                                                 (Map Cluster ([Point], Point)))
       -> Process (Map Cluster ([Point], Point))
    go 0 iteration = do
      iteration (Map.fromList $ map (, cs) segments)
    go n iteration = do
      clusters <- go (n - 1) iteration
      let centers = map snd $ Map.elems clusters
      iteration (Map.fromList $ map (, centers) segments)

    segments :: [(Int, Int)]
    segments = let (lo, _) = bounds points in dividePoints numPoints lo

    dividePoints :: Int -> Int -> [(Int, Int)]
    dividePoints pointsLeft offset
      | pointsLeft <= pointsPerMapper = [(offset, offset + pointsLeft - 1)]
      | otherwise = let offset' = offset + pointsPerMapper in
                    (offset, offset' - 1)
                  : dividePoints (pointsLeft - pointsPerMapper) offset'

    pointsPerMapper :: Int
    pointsPerMapper =
      ceiling (toRational numPoints / toRational (length mappers))

    numPoints :: Int
    numPoints = let (lo, hi) = bounds points in hi - lo + 1