Container Service for Kubernetes:Use Fluid to accelerate data access for Spark applications

Prerequisites

• The ack-spark-operator component is installed. For more information, see Step 1: Install the ack-spark-operator component.

  Note

  The spark.jobNamespaces=["spark"] setting is specified in the Spark job configurations in this topic. If you want to use a different namespace, you must modify the namespace parameter in the Spark job configurations.

• The cloud-native AI suite is installed and the ack-fluid component is deployed. For more information, see Install the cloud-native AI suite.

• Test data is prepared and uploaded to an OSS bucket. For more information, see Prepare and upload test data to an OSS bucket.

Introduction to Fluid

Fluid is an open source, Kubernetes-native orchestrator and accelerator for distributed datasets. Fluid is developed for data-intensive applications in cloud-native scenarios, such as big data applications and AI applications. Key features of Fluid:

• Fluid provides native support for dataset abstraction. This feature provides fundamental support for data-intensive applications, enables efficient data access, and improves the cost-effectiveness of data management in multiple aspects.

• Fluid provides an extensible data engine plug-in with a unified interface for integration with third-party storage services. A variety of runtimes are supported.

• Fluid automates data operations and supports multiple modes to integrate with automated O&M systems.

• Fluid accelerates data access by combining the data caching technology with elastic scaling and data affinity-scheduling.

• Fluid is independent of runtime platforms and supports Kubernetes clusters, Container Service for Kubernetes (ACK) Edge clusters, and ACK Serverless clusters. Fluid is also suitable for multi-cluster scenarios and hybrid cloud scenarios.

For more information about Fluid, see Elastic datasets.

Step 1: Create a dedicated node pool for Fluid

Create a dedicated node pool named fluid for Fluid in your ACK cluster. The node pool is used to deploy the JindoRuntime worker pods in Fluid. The node pool created in this example contains three nodes that are deployed on ecs.d1ne.4xlarge instances, which belong to the network-enhanced big data instance family. The fluid-cloudnative.github.io/node="true" label and the fluid-cloudnative.github.io/node="true":NoSchedule taint are added to each node. Each node is equipped with eight 5,905 GB high-throughput local SATA HDDs, which are formatted and attached to the following directories: /mnt/disk1, /mnt/disk2, ..., and /mnt/disk8. For more information about how to create a node pool, see Create and manage a node pool. For more information about how to select instance types for a node pool, see Best practices for the cache optimization policies of Fluid.

Step 2: Create a dataset

1. Create a Secret YAML file named fluid-oss-secret.yaml to store the credentials used to access your OSS bucket.

   Replace <ACCESS_KEY_ID> and <ACCESS_KEY_SECRET> with an AccessKey pair of your Alibaba Cloud account.

   apiVersion: v1
   kind: Secret
   metadata:
     name: fluid-oss-secret
     namespace: spark
   stringData:
     OSS_ACCESS_KEY_ID: <ACCESS_KEY_ID>
     OSS_ACCESS_KEY_SECRET: <ACCESS_KEY_SECRET>

2. Run the following command to create a Secret:

   kubectl create -f fluid-oss-secret.yaml

   Expected output:

   secret/fluid-oss-secret created

3. Create a YAML file named spark-fluid-dataset.yaml and copy the following content to the file:

   apiVersion: data.fluid.io/v1alpha1
   kind: Dataset
   metadata:
     name: spark
     namespace: spark
   spec:
     mounts:
     - name: spark
       # The OSS path for which you want to accelerate data access. Replace <OSS_BUCKET> with the name of your OSS bucket. 
       mountPoint: oss://<OSS_BUCKET>/
       path: /
       options:
         # The endpoint of the OSS bucket. Replace <OSS_ENDPOINT> with the endpoint of your OSS bucket. 
         # For example, the internal endpoint for OSS buckets in the China (Beijing) region is oss-cn-beijing-internal.aliyuncs.com. 
         fs.oss.endpoint: <OSS_ENDPOINT>
       encryptOptions:
       - name: fs.oss.accessKeyId
         valueFrom:
           secretKeyRef:
             name: fluid-oss-secret
             key: OSS_ACCESS_KEY_ID
       - name: fs.oss.accessKeySecret
         valueFrom:
           secretKeyRef:
             name: fluid-oss-secret
             key: OSS_ACCESS_KEY_SECRET
     # Data will be cached to nodes that match the following affinity rules. 
     nodeAffinity:
       required:
         nodeSelectorTerms:
         - matchExpressions:
           - key: fluid-cloudnative.github.io/node
             operator: In
             values:
             - "true"
     # Tolerations for specific node taints. 
     tolerations:
     - key: fluid-cloudnative.github.io/node
       operator: Equal
       value: "true"
       effect: NoSchedule

   The following list describes the parameters in the preceding code block:

   • mountPoint: the OSS path for which you want to accelerate data access.

   • fs.oss.endpoint: the endpoint of the OSS bucket. For example, the internal endpoint for OSS buckets in the China (Beijing) region is oss-cn-beijing-internal.aliyuncs.com.

   • encryptOptions: retrieves the credentials used to access the OSS bucket from the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET parameters of the fluid-oss-secret Secret.

4. Run the following command to create a dataset:

   kubectl create -f spark-fluid-dataset.yaml

   Expected output:

   dataset.data.fluid.io/spark created

5. Run the following command to query the status of the Dataset:

   kubectl get -n spark dataset spark -o wide

   Expected output:

   NAME    UFS TOTAL SIZE   CACHED   CACHE CAPACITY   CACHED PERCENTAGE   PHASE      HCFS URL   TOTAL FILES   CACHE HIT RATIO   AGE
   spark                                                                  NotBound                                              58m

   The output shows that the dataset is in the NotBound state.

Step 3: Create a JindoRuntime

1. Create a file named spark-fluid-jindoruntime.yaml and copy the following content to the file. The file is used to create a JindoRuntime.

   apiVersion: data.fluid.io/v1alpha1
   kind: JindoRuntime
   metadata:
     # The name must be the same as the name of the dataset you created. 
     name: spark
     namespace: spark
   spec:
     # The number of worker pods. 
     replicas: 3
     tieredstore:
       levels:
       # The cache type is HDD. 
       - mediumtype: HDD
         # The type of the dataset. 
         volumeType: hostPath
         # Set the value based on the number of disks provided by the node. 
         path: /mnt/disk1,/mnt/disk2,/mnt/disk3,/mnt/disk4,/mnt/disk5,/mnt/disk6,/mnt/disk7,/mnt/disk8
         # The cache capacity of each worker pod. 
         quotaList: 5500Gi,5500Gi,5500Gi,5500Gi,5500Gi,5500Gi,5500Gi,5500Gi
         high: "0.99"
         low: "0.95"
     worker:
       resources:
         requests:
           cpu: 14
           memory: 56Gi
         limits:
           cpu: 14
           memory: 56Gi

   The following list describes the parameters in the preceding code block:

   • replicas: the number of worker pods in the JindoFS cluster.

   • mediumtype: the type of the cache.

   • path: the storage path.

   • quota: the maximum capacity of the cache system.

   • high: the upper limit of the storage capacity.

   • low: the lower limit of the storage capacity.

2. Run the following commands to create a JindoRuntime:

   kubectl create -f spark-fluid-jindoruntime.yaml

   Expected output:

   jindoruntime.data.fluid.io/spark created

3. Run the following command to query the status of the JindoRuntime:

   kubectl get -n spark jindoruntime spark

   Expected output:

   NAME    MASTER PHASE   WORKER PHASE   FUSE PHASE   AGE
   spark   Ready          Ready          Ready        2m28s

   In the output, Ready is displayed in the FUSE PHASE column. This indicates that the JindoRuntime is deployed.

4. Run the following command to query the status of the dataset:

   kubectl get -n spark dataset spark -o wide

   Expected output:

   NAME    UFS TOTAL SIZE   CACHED   CACHE CAPACITY   CACHED PERCENTAGE   PHASE   HCFS URL                             TOTAL FILES     CACHE HIT RATIO   AGE
   spark   [Calculating]    0.00B    128.91TiB                            Bound   spark-jindofs-master-0.spark:19434   [Calculating]                     2m5

   The output shows that the dataset is in the Bound state. This indicates that the dataset is deployed.

Step 4: (Optional)Prefetch data

First-time queries cannot hit the cache. To accelerate first-time queries, Fluid provides the DataLoad resource that you can use to prefetch data. Data prefetching preloads data to the cache. This accelerates data access and improves data processing efficiency and system performance.

1. Create a file named spark-fluid-dataload.yaml and copy the following content to the file. The file is used to create a DataLoad.

   apiVersion: data.fluid.io/v1alpha1
   kind: DataLoad
   metadata:
     name: spark
     namespace: spark
   spec:
     dataset:
       name: spark
       namespace: spark
     loadMetadata: true

2. Run the following command to create a DataLoad:

   kubectl create -f spark-fluid-dataload.yaml

   Expected output:

   dataload.data.fluid.io/spark created

3. Run the following command to query the data prefetching progress:

   kubectl get -n spark dataload spark -w

   Expected output:

   NAME    DATASET   PHASE      AGE     DURATION
   spark   spark     Executing   20s   Unfinished
   spark   spark     Complete   9m31s   8m37s

   8m37s is displayed in the DURATION column of the output, which indicates that data prefetching is completed in 8 minutes and 37 seconds.

4. Run the following command to query the status of the dataset:

   kubectl get -n spark dataset spark -o wide

   Expected output:

   NAME    UFS TOTAL SIZE   CACHED      CACHE CAPACITY   CACHED PERCENTAGE   PHASE   HCFS URL                             TOTAL FILES     CACHE HIT RATIO   AGE
   spark   0.00B            326.85GiB   128.91TiB        0.0%                Bound   spark-jindofs-master-0.spark:19434   [Calculating]                     19m

   326.85GiB is displayed in the CACHED column of the output, which indicates that data is preloaded to the cache. 0.00B is displayed in the output of the previous query.

Step 5: Run a Spark job

Step 6: (Optional) Clear the environment

After you complete the preceding steps, you can run the following commands to delete the resources that are no longer required:

kubectl delete -f spark-pagerank-fluid-posix.yaml
kubectl delete -f spark-pagerank-fluid-hcfs.yaml
kubectl delete -f spark-fluid-dataload.yaml
kubectl delete -f spark-fluid-jindoruntime.yaml
kubectl delete -f spark-fluid-dataset.yaml
kubectl delete -f fluid-oss-secret.yaml