Language Detection in Text #4406

assemblies/plugins/pom.xml

@@ -912,6 +912,12 @@
             <version>${project.version}</version>
             <type>zip</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-transform-detectlanguage</artifactId>
+            <version>${project.version}</version>
+            <type>zip</type>
+        </dependency>
         <dependency>
             <groupId>org.apache.hop</groupId>
             <artifactId>hop-transform-detectlastrow</artifactId>

plugins/transforms/detectlanguage/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.hop</groupId>
+        <artifactId>hop-plugins-transforms</artifactId>
+        <version>2.11.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>hop-transform-detectlanguage</artifactId>
+    <packaging>jar</packaging>
+    <name>Hop Plugins Transforms Detect Language</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.github.pemistahl</groupId>
+            <artifactId>lingua</artifactId>
+            <version>1.2.2</version>
+        </dependency>
+        <dependency>
+            <groupId>com.squareup.moshi</groupId>
+            <artifactId>moshi</artifactId>
+            <version>1.3.1</version>
+        </dependency>
+        <dependency>
+            <groupId>it.unimi.dsi</groupId>
+            <artifactId>fastutil</artifactId>
+            <version>8.5.8</version>
+        </dependency>
+    </dependencies>
+</project>

plugins/transforms/detectlanguage/src/assembly/assembly.xml

@@ -0,0 +1,58 @@
+<!--
+  ~ 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.
+  ~
+  -->
+
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.2.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.2.0 http://maven.apache.org/xsd/assembly-2.2.0.xsd">
+    <id>hop-transform-detectlanguage</id>
+    <formats>
+        <format>zip</format>
+    </formats>
+    <baseDirectory>.</baseDirectory>
+    <files>
+        <file>
+            <source>${project.basedir}/src/main/resources/version.xml</source>
+            <outputDirectory>plugins/transforms/detectlanguage</outputDirectory>
+            <filtered>true</filtered>
+        </file>
+    </files>
+
+    <fileSets>
+        <fileSet>
+            <directory>${project.basedir}/src/main/samples</directory>
+            <outputDirectory>config/projects/samples/</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <includes>
+                <include>org.apache.hop:hop-transform-detectlanguage:jar</include>
+            </includes>
+            <outputDirectory>plugins/transforms/detectlanguage</outputDirectory>
+        </dependencySet>
+        <dependencySet>
+            <includes>
+                <include>com.github.pemistahl:lingua:jar</include>
+                <include>it.unimi.dsi:fastutil:jar</include>
+                <include>com.squareup.moshi:moshi:jar</include>
+            </includes>
+            <outputDirectory>plugins/transforms/detectlanguage/lib</outputDirectory>
+        </dependencySet>
+    </dependencySets>
+</assembly>

plugins/transforms/detectlanguage/src/main/java/org/apache/hop/pipeline/transforms/language/DetectLanguage.java

@@ -0,0 +1,253 @@
+/*
+ * 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.hop.pipeline.transforms.language;
+
+import static com.github.pemistahl.lingua.api.Language.UNKNOWN;
+import static com.github.pemistahl.lingua.api.LanguageDetectorBuilder.fromAllSpokenLanguages;
+import static java.lang.System.arraycopy;
+import static java.util.concurrent.ForkJoinPool.commonPool;
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.trim;
+import static org.apache.hop.core.util.Utils.isEmpty;
+
+import com.github.pemistahl.lingua.api.Language;
+import com.github.pemistahl.lingua.api.LanguageDetector;
+import java.util.ArrayList;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+
+public class DetectLanguage extends BaseTransform<DetectLanguageMeta, DetectLanguageData> {
+  private static final Class<?> PKG = DetectLanguage.class;
+
+  public DetectLanguage(
+      TransformMeta transformMeta,
+      DetectLanguageMeta meta,
+      DetectLanguageData data,
+      int copyNr,
+      PipelineMeta pipelineMeta,
+      Pipeline pipeline) {
+    super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
+  }
+
+  public static final int OVER_ALLOCATE_SIZE = 10;
+  private final int cores = Runtime.getRuntime().availableProcessors();
+  private final int maxJobs = cores * 100;
+  private final ExecutorService executor = commonPool();
+  private final AtomicInteger jobs = new AtomicInteger(0);
+
+  private final Object lock = new Object();
+
+  private final LanguageDetector detector =
+      fromAllSpokenLanguages().withPreloadedLanguageModels().build();
+
+  private void finish() {
+    // Wait until all jobs are finished.
+    synchronized (lock) {
+      while (jobs.get() > 0) {
+        try {
+          lock.wait();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+    detector.unloadLanguageModels();
+    setOutputDone();
+  }
+
+  @Override
+  public boolean processRow() throws HopException {
+    Object[] r = getRow(); // Get row from input rowset & set row busy!
+    if (r == null) { // no more input to be expected...
+      finish();
+      return false;
+    }
+
+    if (first) {
+      first = false;
+
+      // get the RowMeta
+      data.previousRowMeta = getInputRowMeta().clone();
+      data.NrPrevFields = data.previousRowMeta.size();
+      data.outputRowMeta = getInputRowMeta().clone();
+      meta.getFields(data.outputRowMeta, getTransformName(), null, null, this, metadataProvider);
+
+      if (isEmpty(meta.getCorpusField())) {
+        throw new HopException(
+            BaseMessages.getString(PKG, "DetectLanguage.Error.CorpusFieldMissing"));
+      }
+      // cache the position of the field
+      cacheIndexPositions();
+    } // End If first
+
+    boolean sendToErrorRow = false;
+    String errorMessage = null;
+
+    process(r);
+
+    try {
+      if (isRowLevel()) {
+        logRowlevel(
+            BaseMessages.getString(
+                PKG,
+                "DetectLanguage.LineNumber",
+                getLinesRead() + " : " + getInputRowMeta().getString(r)));
+      }
+    } catch (Exception e) {
+      if (getTransformMeta().isDoingErrorHandling()) {
+        sendToErrorRow = true;
+        errorMessage = e.toString();
+      } else {
+        logError(
+            BaseMessages.getString(PKG, "DetectLanguage.ErrorInTransformRunning") + e.getMessage());
+        setErrors(1);
+        stopAll();
+        setOutputDone(); // signal end to receiver(s)
+        return false;
+      }
+      if (sendToErrorRow) {
+        // Simply add this row to the error row
+        putError(getInputRowMeta(), r, 1, errorMessage, meta.getCorpusField(), "DetectLanguage001");
+      }
+    }
+
+    return true;
+  }
+
+  private void process(Object[] inputRow) throws HopTransformException {
+    String corpus = trim((String) inputRow[data.indexOfCorpusField]);
+    if (isBlank(corpus)) {
+      return;
+    }
+
+    if (meta.isParallelism()) {
+      processAsync(corpus, inputRow);
+    } else {
+      processSync(corpus, inputRow);
+    }
+  }
+
+  private void processAsync(String corpus, Object[] inputRow) {
+    synchronized (lock) {
+      while (jobs.get() > maxJobs) {
+        try {
+          lock.wait();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+
+    executor.submit(
+        () -> {
+          try {
+            processSync(corpus, inputRow);
+            jobs.decrementAndGet();
+
+            synchronized (lock) {
+              lock.notifyAll();
+            }
+          } catch (Exception e) {
+            logError(e.getMessage(), e);
+          }
+        });
+
+    jobs.incrementAndGet();
+  }
+
+  private void cacheIndexPositions() throws HopException {
+    if (data.indexOfCorpusField < 0) {
+      data.indexOfCorpusField = data.previousRowMeta.indexOfValue(meta.getCorpusField());
+      if (data.indexOfCorpusField < 0) {
+        // The field is unreachable!
+        throw new HopException(
+            BaseMessages.getString(
+                PKG, "DetectLanguage.Exception.CouldnotFindField", meta.getCorpusField()));
+      }
+    }
+    if (data.indexOfDetectedLanguage < 0) {
+      data.indexOfDetectedLanguage = data.outputRowMeta.indexOfValue("detected_language");
+      if (data.indexOfDetectedLanguage < 0) {
+        // The field is unreachable!
+        throw new HopException(
+            BaseMessages.getString(
+                PKG, "DetectLanguage.Exception.CouldnotFindField", "detected_language"));
+      }
+    }
+    if (data.indexOfDetectedLanguageConfidence < 0) {
+      data.indexOfDetectedLanguageConfidence =
+          data.outputRowMeta.indexOfValue("detected_language_confidence");
+      if (data.indexOfDetectedLanguageConfidence < 0) {
+        // The field is unreachable!
+        throw new HopException(
+            BaseMessages.getString(
+                PKG, "DetectLanguage.Exception.CouldnotFindField", "detected_language_confidence"));
+      }
+    }
+  }
+
+  private void processSync(String corpus, Object[] inputRow) throws HopTransformException {
+    SortedMap<Language, Double> confidenceValues = detector.computeLanguageConfidenceValues(corpus);
+    Language detectedLanguage = UNKNOWN;
+    Double confidence = null;
+
+    if (!confidenceValues.isEmpty()) {
+      Language mostLikelyLanguage = confidenceValues.firstKey();
+      Double mostLikelyLanguageProbability = confidenceValues.get(mostLikelyLanguage);
+      if (confidenceValues.size() == 1) {
+        detectedLanguage = mostLikelyLanguage;
+        confidence = mostLikelyLanguageProbability;
+      } else {
+        Double secondMostLikelyLanguageProbability =
+            new ArrayList<>(confidenceValues.values()).get(1);
+        boolean eq =
+            Objects.equals(mostLikelyLanguageProbability, secondMostLikelyLanguageProbability);
+        boolean unk =
+            (mostLikelyLanguageProbability - secondMostLikelyLanguageProbability)
+                < detector.getMinimumRelativeDistance$lingua();
+        if (!eq && !unk) {
+          detectedLanguage = mostLikelyLanguage;
+          confidence = mostLikelyLanguageProbability;
+        }
+      }
+    }
+
+    Object[] outputRow = new Object[inputRow.length + 2 + OVER_ALLOCATE_SIZE];
+    arraycopy(inputRow, 0, outputRow, 0, inputRow.length);
+    outputRow[data.indexOfDetectedLanguage] = detectedLanguage.name();
+    outputRow[data.indexOfDetectedLanguageConfidence] = confidence;
+    putRow(data.outputRowMeta, outputRow);
+  }
+
+  @Override
+  public void putRow(IRowMeta rowMeta, Object[] row) throws HopTransformException {
+    synchronized (this) {
+      super.putRow(rowMeta, row);
+    }
+  }
+}