MaxCompute:Diagnostic cases of LogView

Diagnose a job that fails to run

If a job fails to run, you can view error information on the Result tab of LogView. The Result tab is automatically displayed when you open LogView for a failed job.

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 execute an INSERT INTO statement 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

  Problem description: Task rerun appears multiple times on the SubStatusHistory tab, and FAILED: ODPS-0110141:Data version exception appears on the Result tab. In this case, the job does not fail. Instead, the job 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

An execution plan is displayed on the Job Details tab of LogView. The execution plan is not complete and the job is in the Running state. If a job is stuck at the execution stage or requires an unexpectedly long period of time to complete at this stage, this issue may occur due to the following reasons: waiting for resources, data skew, inefficient UDF execution, and data bloat. This section describes the characteristics in each case and the solutions.

• Waiting for resources

  Symptom: An instance is in the Ready state, or some instances are in the Running state while others are in the Ready state. Note that if an instance is in the Ready state but has a Debug history entry, it may indicate a retry triggered by an instance failure rather than waiting for resources. In the Fuxi instance list, you can use SmartFilter to filter by status. The Status column shows the current state of each instance.

  Solution:

  1. Determine if the queuing status is normal. You can check the job's position in the queue by viewing the Queue length in LogView. The Queue length value is available 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 MaxCompute console, select Quota management > Resource usage from the left-side menu and view the trend graphs for metrics like CPU resource for the target quota group. If the usage of a resource is near or over its quota, it indicates that the quota group is low on resources, and queuing is expected. The scheduling order of a job depends not only on its submission time and priority but also on whether its required memory or CPU resources can be met.

  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 taking much longer to finish. These slow instances might be processing more data than others. In SmartFilter, you can check the Long-tails count and compare instance latencies 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 method: If a task fails, you can examine the DAG on the Job details tab in LogView to determine if it contains a UDF. In the DAG, a failed task such as R4_3 appears in red. The fx label on the node indicates the UDF language, such as ["JAVA"]. Double-click R4_3 to open the operator view, which shows the operator chain from top to bottom and lists all UDF names used in the task. In the task's StdOut log, the UDF framework prints the number of input records, output records, and processing time. You can use this data to identify performance issues. Normally, the Speed(records/s) is in the range of hundreds of thousands to millions. If it drops to the tens of thousands, a performance problem is likely. The log displays a summary of UDF processing and a statistical table for each operator (CursorId) with metrics like OutputCount, InnerTime, and Speed(records/s).

  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: The amount of output data from a task is much larger than the amount of input data. For example, if 1 GB of data is processed and becomes 1 TB, the performance is significantly reduced when a single instance processes the 1 TB of data. After a job is complete, you can view the input and output data volumes in the I/O Records of the task. If a job gets stuck in the Join stage, you can check the StdOut logs of a few Fuxi instances that are in the Running state. In SmartFilter, filter for instances in the Running state, and then click the StdOut icon of an instance to view its logs. If the StdOut log continuously shows Merge Join logs, it indicates that a single worker is continuously performing a Merge Join. If the number of output rows from the Merge Join exceeds 143.3 billion, this indicates severe data inflation, and you need to check whether the JOIN condition and Join Key are appropriate. The log will continuously output Merge join cursor records, and the value of the OutputRowCount field will keep increasing. You can use this to determine the degree of data inflation.

  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 runs first. Its result is needed for partition pruning to optimize the number of partitions the main query reads. These two runs are separate Fuxi jobs. LogView displays each Fuxi job in a separate tab, such as SQL_0_0_0_job_0 and SQL_0_0_1_job_1. The DAG also shows the dependency between multiple JOB nodes. Simply click the second tab to see the execution status of job_1. After switching, you can see the current status of each task in job_1, such as Running or Waiting.

• 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 use LogView to check whether a job has triggered an automatic small file merge. Similar to the multi-stage execution of subqueries, the Merge job is also displayed as a separate tab. Although the additional MergeTask for the automatic small file merge increases the overall execution time of the current job, it optimizes the number and size of files generated in the result table after the merge. This prevents excessive pressure on the file system and improves the read performance when the table is used by subsequent jobs. The tab name of the Merge job is similar to SQL_0_0_0_merge, which displays the execution progress of the MergeTask.

  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 finishes, it still takes some time to update the table's metadata. LogView's SubStatusHistory shows that the job is stuck at SQLTask is updating meta information. The corresponding status code is 1260, and its Latency value shows how long the metadata update took.

• Increase in the size of the output file

  Symptom: The number of input and output records is similar, but the output size is several times larger. In Fuxi Jobs, check the IO bytes column to compare the Input and Output data sizes for each task. This helps identify the stage where the size inflation occurred.

  Solution: One reason is a change in data distribution. When writing data to a table, the data is compressed. The compression algorithm is most effective on repetitive data. Therefore, if identical data is arranged together during the write process, a high compression ratio can be achieved. The data distribution depends on how data is shuffled and sorted during the write stage (the R12 stage in Fuxi Jobs). In this example, the final SQL operation is a JOIN, with the JOIN key being:

  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.