[
https://issues.apache.org/jira/browse/KYLIN-5097?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17466300#comment-17466300
]
ASF GitHub Bot commented on KYLIN-5097:
---------------------------------------
zhangayqian commented on a change in pull request #1778:
URL: https://github.com/apache/kylin/pull/1778#discussion_r776139094
##########
File path:
source-hive/src/main/java/org/apache/kylin/source/hive/CreateSparkHiveDictStep.java
##########
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kylin.source.hive;
+
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionBuilder;
+import org.apache.commons.cli.Options;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.lock.DistributedLock;
+import org.apache.kylin.common.util.AbstractApplication;
+import org.apache.kylin.common.util.OptionsHelper;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.engine.EngineFactory;
+import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
+import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.engine.mr.common.SerializableConfiguration;
+import org.apache.kylin.job.common.PatternedLogger;
+import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
+import org.apache.kylin.shaded.com.google.common.collect.Lists;
+import org.apache.kylin.shaded.com.google.common.collect.Maps;
+import org.apache.spark.TaskContext;
+import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class CreateSparkHiveDictStep extends AbstractApplication implements
Serializable {
+
+ private static final Logger logger =
LoggerFactory.getLogger(CreateSparkHiveDictStep.class);
+
+ private transient final PatternedLogger stepLogger = new
PatternedLogger(logger);
+
+ public static final Option OPTION_CUBE_NAME =
OptionBuilder.withArgName(BatchConstants.ARG_CUBE_NAME).hasArg().isRequired(true).withDescription("Cube
Name").create(BatchConstants.ARG_CUBE_NAME);
+ public static final Option OPTION_HIVE_DICT_COLUMNS =
OptionBuilder.withArgName("hiveDictColumns").hasArg().isRequired(true).withDescription("Hive
Dict Columns").create("hiveDictColumns");
+ public static final Option OPTION_META_URL =
OptionBuilder.withArgName("metaUrl").hasArg().isRequired(true).withDescription("Meta
Url").create("metaUrl");
+ public static final Option OPTION_SEGMENT_ID =
OptionBuilder.withArgName("segmentId").hasArg().isRequired(true).withDescription("Segment
Id").create("segmentId");
+ public static final Option OPTION_FLOW_JOB_ID =
OptionBuilder.withArgName("flowJobId").hasArg().isRequired(true).withDescription("Flow
Job Id").create("flowJobId");
+ public static final Option OPTION_WAREHOUSE_DIR =
OptionBuilder.withArgName("wareHouseDir").hasArg().isRequired(true).withDescription("WareHouse
Dir").create("wareHouseDir");
+
+ private final Lock threadLock = new ReentrantLock();
+
+ private Options options;
+
+ public CreateSparkHiveDictStep() {
+ options = new Options();
+ options.addOption(OPTION_CUBE_NAME);
+ options.addOption(OPTION_HIVE_DICT_COLUMNS);
+ options.addOption(OPTION_META_URL);
+ options.addOption(OPTION_SEGMENT_ID);
+ options.addOption(OPTION_FLOW_JOB_ID);
+ options.addOption(OPTION_WAREHOUSE_DIR);
+ }
+
+ private String cubeName;
+ private String segmentId;
+ private String[] hiveDictColumns;
+ private String metaUrl;
+ private String flowJobId;
+ private String wareHouseDir;
+
+ private void initParam(OptionsHelper optionsHelper) {
+ cubeName = optionsHelper.getOptionValue(OPTION_CUBE_NAME);
+ segmentId = optionsHelper.getOptionValue(OPTION_SEGMENT_ID);
+ hiveDictColumns =
optionsHelper.getOptionValue(OPTION_HIVE_DICT_COLUMNS).split(",");
+ metaUrl = optionsHelper.getOptionValue(OPTION_META_URL);
+ flowJobId = optionsHelper.getOptionValue(OPTION_FLOW_JOB_ID);
+ wareHouseDir = optionsHelper.getOptionValue(OPTION_WAREHOUSE_DIR);
+ }
+
+ @Override
+ protected void execute(OptionsHelper optionsHelper) {
+ initParam(optionsHelper);
+
+ DistributedLock lock = null;
+ try {
+ lock =
KylinConfig.getInstanceFromEnv().getDistributedLockFactory().lockForCurrentThread();
+
+ createSparkHiveDict(lock);
+
+ } catch (Exception e) {
+ logger.error("", e);
Review comment:
Why not throw this exception?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@kylin.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> modify hive dict table format from textfile to orc
> --------------------------------------------------
>
> Key: KYLIN-5097
> URL: https://issues.apache.org/jira/browse/KYLIN-5097
> Project: Kylin
> Issue Type: Bug
> Affects Versions: v3.1.1, v3.1.2, v3.1.3
> Reporter: fengpod
> Priority: Minor
>
> Textfile format end with ‘\n’ by default， when the data itself contains ‘\n’,
> dict table may appear dirty data。
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
