<commit_before>/**
  * (C) Copyright IBM Corp. 2015 - 2017
  *
  * Licensed 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 com.ibm.sparktc.sparkbench.utils

import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SparkSession}
import com.databricks.spark.avro._
import org.apache.hadoop.fs.FileSystem

object SparkFuncs {

  private def parseFormat(outputDir: String, fileFormat: Option[String]): String = (outputDir, fileFormat) match {
    case (Formats.console, None) => Formats.console
    case (_, None)         => outputDir.split('.').last
    case (_, Some(s))      => s
  }

  def verifyOutput(outputDir: Option[String], saveMode: String, spark: SparkSession, fileFormat: Option[String] = None): Unit = {
    verifyCanWriteOrThrow(outputDir, saveMode, spark, fileFormat)
    verifyFormatOrThrow(outputDir, fileFormat)
  }
  
  def verifyCanWrite(outputDir: String, saveMode: String, spark: SparkSession, fileFormat: Option[String] = None): Boolean = {
    if(outputDir == Formats.console) true
    else {
      (pathExists(outputDir, spark), saveMode) match {
        case (false, _) => true
        case (true, SaveModes.overwrite) => true
        case (true, SaveModes.append) => true
        case (true, SaveModes.error) => false
        case (true, SaveModes.ignore) => true // allow write operation to no-op when the time comes
        case (true, _) => throw SparkBenchException(s"Unrecognized or unimplemented save-mode: $saveMode")
      }
    }
  }

  def verifyFormat(outputDir: String, fileFormat: Option[String] = None): Boolean = {
    val format = parseFormat(outputDir, fileFormat)

    format match {
      case Formats.parquet => true
      case Formats.csv => true
      case Formats.orc => true
      case Formats.avro => true
      case Formats.json => true
      case Formats.console => true
      case _ => false
    }
  }

  private def extractHDFSuri(str: String): Option[String] = {
    val regx = """hdfs://(hdfs:\d+)""".r.unanchored
    str match {
      case regx(uri) => Some(uri)
      case _ => None
    }
  }

  private def getHadoopFS(path: String, spark: SparkSession): FileSystem = {
    val uriOpt = extractHDFSuri(path)
    if(uriOpt.nonEmpty){
      org.apache.hadoop.fs.FileSystem.get(new java.net.URI(s"hdfs://${uriOpt.get}"), new org.apache.hadoop.conf.Configuration())
    }
    else {
      val conf = spark.sparkContext.hadoopConfiguration
      org.apache.hadoop.fs.FileSystem.get(conf)
    }
  }

  def pathExists(path: String, spark: SparkSession): Boolean = {
    val fs: FileSystem = getHadoopFS(path, spark)
    fs.exists(new org.apache.hadoop.fs.Path(path))
  }

  def verifyPathExistsOrThrow(path: String, errorMessage: String, spark: SparkSession): String = {
    if (pathExists(path, spark)) { path } // "It is true that this file exists"
    else { throw SparkBenchException(errorMessage) }
  }

  def verifyPathNotExistsOrThrow(path: String, errorMessage: String, spark: SparkSession): String = {
    if (!pathExists(path, spark)) { path } // "It is true that this file does not exist"
    else { throw SparkBenchException(errorMessage) }
  }<commit_msg>Fixing regex for hdfs urls<commit_after>/**
  * (C) Copyright IBM Corp. 2015 - 2017
  *
  * Licensed 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 com.ibm.sparktc.sparkbench.utils

import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SparkSession}
import com.databricks.spark.avro._
import org.apache.hadoop.fs.FileSystem

object SparkFuncs {

  private def parseFormat(outputDir: String, fileFormat: Option[String]): String = (outputDir, fileFormat) match {
    case (Formats.console, None) => Formats.console
    case (_, None)         => outputDir.split('.').last
    case (_, Some(s))      => s
  }

  def verifyOutput(outputDir: Option[String], saveMode: String, spark: SparkSession, fileFormat: Option[String] = None): Unit = {
    verifyCanWriteOrThrow(outputDir, saveMode, spark, fileFormat)
    verifyFormatOrThrow(outputDir, fileFormat)
  }
  
  def verifyCanWrite(outputDir: String, saveMode: String, spark: SparkSession, fileFormat: Option[String] = None): Boolean = {
    if(outputDir == Formats.console) true
    else {
      (pathExists(outputDir, spark), saveMode) match {
        case (false, _) => true
        case (true, SaveModes.overwrite) => true
        case (true, SaveModes.append) => true
        case (true, SaveModes.error) => false
        case (true, SaveModes.ignore) => true // allow write operation to no-op when the time comes
        case (true, _) => throw SparkBenchException(s"Unrecognized or unimplemented save-mode: $saveMode")
      }
    }
  }

  def verifyFormat(outputDir: String, fileFormat: Option[String] = None): Boolean = {
    val format = parseFormat(outputDir, fileFormat)

    format match {
      case Formats.parquet => true
      case Formats.csv => true
      case Formats.orc => true
      case Formats.avro => true
      case Formats.json => true
      case Formats.console => true
      case _ => false
    }
  }

  private def extractHDFSuri(str: String): Option[String] = {
    val regx = """hdfs://(\S+:\d+)""".r.unanchored
    str match {
      case regx(uri) => Some(uri)
      case _ => None
    }
  }

  private def getHadoopFS(path: String, spark: SparkSession): FileSystem = {
    val uriOpt = extractHDFSuri(path)
    if(uriOpt.nonEmpty){
      org.apache.hadoop.fs.FileSystem.get(new java.net.URI(s"hdfs://${uriOpt.get}"), new org.apache.hadoop.conf.Configuration())
    }
    else {
      val conf = spark.sparkContext.hadoopConfiguration
      org.apache.hadoop.fs.FileSystem.get(conf)
    }
  }

  def pathExists(path: String, spark: SparkSession): Boolean = {
    val fs: FileSystem = getHadoopFS(path, spark)
    fs.exists(new org.apache.hadoop.fs.Path(path))
  }

  def verifyPathExistsOrThrow(path: String, errorMessage: String, spark: SparkSession): String = {
    if (pathExists(path, spark)) { path } // "It is true that this file exists"
    else { throw SparkBenchException(errorMessage) }
  }

  def verifyPathNotExistsOrThrow(path: String, errorMessage: String, spark: SparkSession): String = {
    if (!pathExists(path, spark)) { path } // "It is true that this file does not exist"
    else { throw SparkBenchException(errorMessage) }
  }
{'fraction_non_alphanumeric': 0.11513336962438758, 'fraction_numerical': 0.00326619488296135, 'mean_word_length': 3.369203329369798, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6962288', 'n_tokens_mistral': 2291, 'n_tokens_neox': 2197, 'n_words': 638}