MaxCompute:Diagnostic cases of LogView

Diagnose a job that fails to run

When a job fails, you can view the error information on the result tab in LogView. For a failed job, LogView opens to the result tab by default.

Diagnose a job that runs slowly

Compilation stage

• Scheduling

  Problem description: The sub-status of the job is Waiting for cluster resource. The job is waiting to be compiled.

  Cause: The resources of the computing cluster are insufficient.

  Solution: Check the status of the computing cluster and the required resources of the computing cluster. If you use a subscription cluster, you can scale out resources.

• Optimization

  Problem description: The sub-status of the job is SQLTask is optimizing query. The optimizer is optimizing the execution plan.

  Cause: The execution plan is complex. The optimizer requires a long period of time to optimize the execution plan.

  Solution: Wait for the optimizer to complete the optimization. This process takes less than 10 minutes in most cases.

• Generation of an execution plan

  Problem description: The sub-status of the job is SQLTask is generating execution plan.

  Cause 1: Data of an excessive number of partitions is read.

  Solution: Optimize SQL statements to reduce the number of partitions. For example, you can perform partition pruning, filter out partitions from which data does not need to be read, and split large jobs into small jobs. For more information about how to determine whether partition pruning takes effect in SQL statements and common scenarios in which partition pruning fails, see Check whether partition pruning is effective.

  Cause 2: Excessive small files are generated. Small files are generated in the following scenarios:

  1. An incorrect operation is performed when Tunnel commands are used to upload data. For example, a new upload session is created each time a data record is uploaded. For more information, see FAQ about Tunnel commands.

  2. When you run an insert into operation on a partitioned table, a new file is generated in the partition directory.

  Solutions

  1. Use the TunnelBufferedWriter interface to upload data in a more efficient manner. This prevents excessive small files from being generated.

  2. Manually merge small files. For more information, see Merge small files.

  Note

  If the number of small files is greater than 10,000, you can enable automatic merging of small files. The system automatically merges small files every day. However, if the system fails to merge small files in special scenarios, you must manually merge the small files.

• Data replication across clusters

  Symptom: The sub-status list shows multiple instances of Task rerun, and the Result tab displays the error message FAILED: ODPS-0110141:Data version exception. The job may seem to have failed but is actually still running, which indicates that it is replicating data across clusters.

  Cause 1: Data is migrated between clusters for the project. In this case, a large number of jobs that replicate data across clusters are running in the first one or two days after the migration is complete.

  Solution: Wait until data replication across clusters is complete as expected.

  Cause 2: Data is migrated between clusters for the project. However, partitions are not filtered as expected. As a result, old data is read from specific partitions.

  Solution: Filter out the partitions that contain old data.

Execution stage

The job details page in LogView shows an incomplete execution plan, and the job status is still Running. During the execution stage, a job might get stuck or run longer than expected due to resource waits, data skew, inefficient UDF execution, or data bloat. The following sections describe the symptoms and solutions for each scenario.

• Waiting for resources

  Symptom: Instances are in the Ready state, or some are Running while others are Ready. Note that if an instance is Ready but has a Debug history, it might be retrying after a failure, not waiting for resources. In the Fuxi instance list, you can filter by status using SmartFilter. The Status column shows the current state of each instance.

  Solution:

  1. Determine if the queuing is normal. You can check the job's position in the queue by looking at the Queue Length in the Basic Info panel on the left side of LogView. Alternatively, check the resource usage of the corresponding quota group in the MaxCompute console. In the left-side navigation pane of the MaxCompute console, choose Project Management > Quota Management > Resource Consumption to view trend charts for metrics like CPU Resource for the target quota group. If the usage of a resource is close to or exceeds its quota, the quota group is experiencing a resource shortage, and job queuing is normal. The scheduling order depends not only on the submission time and priority but also on whether the required memory or CPU resources are available.

  2. View the jobs that use the quota group.

     Large jobs that have low priorities may be submitted, or multiple small jobs are submitted at a time. The jobs occupy a large number of resources. You can contact the owner of the jobs to terminate the jobs and release the resources that are occupied by the jobs.

  3. Change the quota group of the job to a quota group of another project.

  4. Scale out resources. This solution is suitable only for users who use subscription resources.

• Data skew

  Symptom: Most instances in a task have finished, but a few "long-tail" instances are still running. These instances are running slowly, possibly because they are processing a larger amount of data. In SmartFilter, you can check the Long-Tails count and compare the duration of each instance in the Latency column.

  Solution: For more information about common causes of data skew and the related optimization methods, see Data skew tuning.

• Inefficient UDF execution

  In this topic, UDFs refer to various user-defined extensions, including user-defined scalar functions (UDFs), user-defined aggregate functions (UDAFs), user-defined table-valued functions (UDTFs), user-defined joins (UDJs), and user-defined types (UDTs).

  Description: The execution efficiency of a task is low, and the task includes UDFs. The following error message that indicates a UDF execution timeout may appear: Fuxi job failed - WorkerRestart errCode:252,errMsg:kInstanceMonitorTimeout, usually caused by bad udf performance.

  Troubleshooting: When a task fails, you can quickly determine if it contains a UDF by examining the DAG on the job details tab in LogView. In the DAG, a failed task is colored red, and an fx label such as ["JAVA"] on the node indicates the UDF's language. For example, you might see that the failed task R4_3 contains a UDF written in Java. Double-click R4_3 to open the Operator view, which lists all UDFs in that task. The Operator view displays the operator chain from top to bottom, with a list of all referenced UDFs at the bottom. Additionally, the StdOut log for the task shows the UDF's input record count, output record count, and processing time. This data helps you identify performance issues. Typically, a Speed(records/s) of hundreds of thousands or millions is normal. If it drops to the tens of thousands, a performance problem is likely. The log displays a summary of UDF processing and a statistics table with OutputCount, InnerTime, and Speed(records/s) for each operator (CursorId).

  Solution: If a performance issue occurs, you can use the following method to troubleshoot the issue and optimize the performance.

  1. Check whether an error occurs in the UDF.

     In specific cases, the performance issue is caused by a specific data value. For example, an infinite loop occurs when a specific value appears. MaxCompute Studio allows you to download specific sample data of a table and use the data on your on-premises machine for troubleshooting. For more information, see Java UDFs and Python UDF in the MaxCompute Studio development manual.

  2. Check whether the name of the UDF is the same as a built-in function.

     A built-in function may be overwritten by a UDF whose name is the same. If a function appears to be a built-in function, you must determine whether a UDF whose name is the same can overwrite the built-in function.

  3. Use a built-in function to replace the UDF.

     If built-in functions that provide similar features exist, we recommend that you do not use UDFs. Built-in functions are verified and used in a more efficient manner. In addition, optimizers perform white-box testing on built-in functions. Therefore, more optimizations can be made. For more information about the usage of built-in functions, see Built-in functions.

  4. Replace specific UDFs with the built-in functions that provide similar features, and retain only the UDFs that cannot be implemented by using built-in functions.

  5. Optimize the evaluate method of UDFs.

     Use the evaluate method in only necessary operations related to parameters. Perform initialization-related operations or repeated calculations in advance because the evaluate method is repeatedly run.

  6. Estimate the period of time that is required to run a UDF.

     Simulate the amount of data that is processed by one instance on your on-promises machine to test the period of time that is required to run a UDF. Then, optimize the implementation of the UDF. By default, the maximum period of time that is required to run a UDF is 30 minutes. A UDF must return data within 30 minutes or use context.progress() to report heartbeats. If the estimated period of time that is required to run a UDF is longer than 30 minutes, you can configure a parameter to specify the timeout period of a UDF.

     Default value: 1800. Unit: seconds. Valid values: 1 to 3600.
     -- Specify the timeout period of a UDF. Unit: seconds. Default value: 600.
     -- You can manually adjust the timeout period of the UDF in the range of [0,3600].

  7. Modify memory parameters.

     The low efficiency of UDFs is not necessarily due to the computational complexity. The storage complexity may also affect the efficiency of UDFs. Examples:

     • Memory overflow occurs if a UDF performs memory computing on or sorts a large amount of data.

     • Insufficient memory causes high garbage collection (GC) frequency.

     You can modify memory parameters to temporarily handle the preceding issues. The specific optimization must be performed based on your business requirements. Example:

     set odps.sql.udf.jvm.memory=
     -- Specify the maximum memory size that can be used for the JVM heap of a UDF. Default value: 1024. Unit: MB.
     -- You can change the value of the odps.sql.udf.jvm.memory parameter in the range of [256,12288].

     Note

     If a UDF is used, partition pruning may become invalid. The UdfProperty annotation is supported from MaxCompute V2.0. When you define a UDF, you can use the annotation for the compiler to specify that the UDF is deterministic. Sample code:

     @com.aliyun.odps.udf.annotation.UdfProperty(isDeterministic = true)
     public class AnnotatedUdf extends com.aliyun.odps.udf.UDF {
         public String evaluate(String x) {
             return x;
         }
     }

     If you rewrite the SQL statement for a UDF, you can use the UDF in partition filtering.

     -- SQL statement before rewriting
     SELECT * FROM t WHERE pt = udf('foo');  -- pt indicates a partition key column of t.
     -- SQL statement after rewriting
     SELECT * FROM t WHERE pt = (SELECT udf('foo')); --pt indicates a partition key column of t.

• Data bloat

  Symptom: A task's output data is much larger than its input data. For example, if 1 GB of data becomes 1 TB after processing, the performance of an instance handling 1 TB of data will be severely degraded. After the job completes, the I/O Records of the task show the input and output data volumes. If a job is stuck in the Join stage, select a few Running instances using SmartFilter and check their StdOut logs by clicking the StdOut icon. If the StdOut log continuously shows Merge join cursor records and the OutputRowCount value keeps increasing, it indicates severe data bloat. You need to check if the JOIN condition and join key are reasonable.

  Solution:

  • Check the following issues in the code: whether the JOIN condition is correct, whether the JOIN condition is written as a Cartesian product, whether the UDTF is normal, and whether excessive output data is generated.

  • Check whether data bloat is caused by aggregation.

    Most aggregators perform recursive aggregation. When an aggregator aggregates data, the aggregator first merges the intermediate results. The amount of the intermediate result data is not large and the computing complexity of most aggregators is low. Aggregation can be complete quickly even if the amount of data is large. In most cases, data bloat does not occur during aggregation. However, if you perform aggregation in the following scenarios, data bloat may occur:

    • Enable aggregation in the SELECT statement to perform the DISTINCT operation in different dimensions. Data expands each time the DISTINCT operation is performed.

    • If you use the GROUPING SETS or CUBE | ROLLUP statement, the size of the intermediate result data may be increased multiple times compared with the original data size. However, if you use a specific statement such as COLLECT_LIST or MEDIAN, you must retain all intermediate result data. This may cause specific issues.

  • Prevent data bloat that occurs due to JOIN operations.

    For example, you want to join two tables. The left table contains a large amount of population data. However, the processing efficiency is high because of the high parallelism of MaxCompute instances. The right table is a dimension table that records information about each gender, such as possible bad habits of each gender. The dimension table contains only two genders but hundreds of rows that correspond to each gender exist. If you join the tables by gender, data in the left table may be expanded hundreds of times. To prevent data bloat, you can aggregate the rows of data in the right table into two rows of data before you join the tables.

  • Check whether data bloat is caused by the GROUPING SET statement. When the GROUPING SET statement is executed, data is expanded and the output data is increased multiple times compared with the number of groups. The current execution plan cannot adapt to the GROUPING SET statement or change the degree of parallelism of downstream tasks. You can manually configure the degree of parallelism of downstream tasks. Sample statements:

    set odps.stage.reducer.num = xxx; 
    set odps.stage.joiner.num = xxx;

Termination stage

• Subquery execution at multiple stages

  In most cases, subqueries of MaxCompute SQL are compiled into the same Fuxi DAG. Therefore, all subqueries and main queries are complete by one Fuxi job. However, specific special subqueries need to be separately run before main queries. Sample code:

  SELECT
       product,
      sum(price)
   FROM
       sales
   WHERE
       ds in (SELECT DISTINCT ds FROM t_ds_set)
   GROUP BY product;

  The subquery SELECT DISTINCT ds FROM t_ds_set is executed first, and its result is needed for partition pruning to optimize the number of partitions the main query reads. These two runs are separate Fuxi jobs. In LogView, each Fuxi job appears in a separate tab, such as SQL_0_0_0_job_0 and SQL_0_0_1_job_1. The DAG also shows the dependencies between multiple JOB nodes. Simply click the second tab to view the execution status of job_1. After switching, you can see the current status (such as Running or Waiting) of each task in job_1.

• Excessive small files

  Excessive small files mainly affect the storage performance and computing performance.

  • Storage: Excessive small files increase the workload on the Apsara Distributed File System. This affects the storage usage.

  • Computing: The overall processing performance is affected because the processing efficiency of MaxCompute on a single large file is higher than the processing efficiency on multiple small files. Therefore, when an SQL job is complete, the operation that is used to merge small files is automatically triggered if specific conditions are met. This helps prevent the system from generating excessive small files.

• Solution: You can check in LogView if the job triggered an automatic small file merge. Similar to multi-stage subquery execution, the merge job appears in a separate tab. A merge job's tab name might look like SQL_0_0_0_merge and shows the progress of the MergeTask. Although this adds to the total job execution time, it results in a more reasonable number and size of files in the output table. This prevents excessive pressure on the file system and improves read performance for subsequent jobs that use the table.

  If excessive small files exist, SELECT statements may be executed for a long period of time when a job is at the termination stage. When the system generates and displays the execution results of SELECT statements, the system needs to open a large number of small files to read data from the files. This process is time-consuming. To prevent the system from generating a large number of execution results, we recommend that you do not use SELECT statements. You can use Tunnel commands to download data. If the number of execution results is not large but the number of files is excessively large, we recommend that you check whether the odps.merge.smallfile.filesize.threshold parameter is properly configured. For more information about how to merge small files, see Merge small files.

• Metadata update in dynamic partitions

  Problem description: After a Fuxi job is complete, you may need to perform specific operations related to the metadata. For example, if you want to move the result data to a specific directory and update the metadata of the table, a large number of partitions may be generated in the table during dynamic partitioning. As a result, the process is time-consuming. For example, the insert into ... values statement is executed for the partitioned table sales to add 2,000 partitions. Sample statement:

  INSERT INTO TABLE sales partition (ds)(ds, product, price)
  VALUES ('20170101','a',1),('20170102','b',2),('20170103','c',3), ...;

  After the Fuxi job completes, it still takes time to update the table's metadata. LogView's SubStatusHistory shows that the job is stuck at SQLTask is updating meta information. This corresponds to status code 1260, and you can check its Latency value to see how much time the metadata update took.

• Increase in the size of the output file

  Symptom: Data may expand by several times even if the numbers of input and output records are similar. In the IO Bytes column of Fuxi Jobs, you can compare the Input and Output data sizes of each Task to identify the stage at which the data expansion occurred.

  Solution: One cause is a change in data distribution, which affects compression. During the process of writing data to a table, data is compressed. Compression algorithms achieve the highest compression ratio on repetitive data. Therefore, if identical data is grouped together during the writing process, a high compression ratio can be achieved. Data distribution largely depends on how data is shuffled and sorted during the write stage (R12 in Fuxi Jobs). In this example, the final SQL operation is a JOIN, with the following join key:

  on  t1.query = t2.query and t1.item_id=t2.item_id

  The characteristics of the data indicate that most columns are the attributes of items. In this case, all columns for the items that have the same item_id are the same. Therefore, all items in the query are sorted in random order. This causes a low compression ratio. The following sample code provides an example on how to modify the join sequence. After the modification, the data size is decreased to 1/3 of the original data size.

  on t1.item_id=t2.item_id and t1.query = t2.query

  After the modification, the data size is decreased to 1/3 of the original data size.

  In another case, the shuffle operations that are generated by JOIN or GROUP BY do not contain the most suitable sorting column for compression. In this case, you can use the ZORDER BY clause to sort the items on your on-premises machine. This way, you can obtain a high compression ratio at a lower cost. You can also execute a DISTRIBUTED BY SORT BY statement to manually rearrange the data distribution. This method requires a long period of time for data calculation and a high CPU utilization.