Flink, I am able to access Kinesis from Intellij but not S3 I have edited my stack overflow question with kinesis code , Flink is still having issues reading S3.
https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117656862_66536868 Thanks Sri On Tue, Mar 9, 2021 at 11:30 AM sri hari kali charan Tummala < kali.tumm...@gmail.com> wrote: > my stack overflow question. > > > https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868 > > On Tue, Mar 9, 2021 at 11:28 AM sri hari kali charan Tummala < > kali.tumm...@gmail.com> wrote: > >> Here is my Intellij question. >> >> >> https://stackoverflow.com/questions/66536868/flink-aws-s3-access-issue-intellij-idea?noredirect=1#comment117626682_66536868 >> >> On Mon, Mar 8, 2021 at 11:22 AM sri hari kali charan Tummala < >> kali.tumm...@gmail.com> wrote: >> >>> >>> Hi Flink Experts, >>>> >>> >>> I am trying to read an S3 file from my Intellij using Flink I am.comimg >>>> across Aws Auth error can someone help below are all the details. >>>> >>> >>> >>>> I have Aws credentials in homefolder/.aws/credentials >>>> >>> >>> My Intellij Environment Variables:- >>>> ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.8.1 >>>> >>>> FLINK_CONF_DIR=/Users/Documents/FlinkStreamAndSql-master/src/main/resources/flink-config >>>> >>>> flink-conf.yaml file content:- >>>> >>>> fs.hdfs.hadoopconf: >>>> /Users/blah/Documents/FlinkStreamAndSql-master/src/main/resources/hadoop-config >>>> >>>> core-site.xml file content:- >>>> >>>> <?xml version="1.0"?> >>>> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> >>>> >>>> <configuration> >>>> <property> >>>> <name>fs.s3.impl</name> >>>> <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value> >>>> </property> >>>> >>>> <property> >>>> <name>fs.s3.buffer.dir</name> >>>> <value>/tmp</value> >>>> </property> >>>> >>>> <property> >>>> <name>fs.s3a.server-side-encryption-algorithm</name> >>>> <value>AES256</value> >>>> </property> >>>> >>>> <!--<property> >>>> <name>fs.s3a.aws.credentials.provider</name> >>>> >>>> <value>org.apache.hadoop.fs.s3a.SharedInstanceProfileCredentialsProvider</value> >>>> </property>--> >>>> >>>> <property> >>>> <name>fs.s3a.aws.credentials.provider</name> >>>> >>>> <value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.access.key</name> >>>> <value></value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.secret.key</name> >>>> <value></value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.session.token</name> >>>> <value></value> >>>> </property> >>>> >>>> <property> >>>> <name>fs.s3a.proxy.host</name> >>>> <value></value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.proxy.port</name> >>>> <value>8099</value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.proxy.username</name> >>>> <value></value> >>>> </property> >>>> <property> >>>> <name>fs.s3a.proxy.password</name> >>>> <value></value> >>>> </property> >>>> >>>> </configuration> >>>> >>>> POM.xml file:- >>>> >>>> <?xml version="1.0" encoding="UTF-8"?> >>>> <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> >>>> >>>> <groupId>FlinkStreamAndSql</groupId> >>>> <artifactId>FlinkStreamAndSql</artifactId> >>>> <version>1.0-SNAPSHOT</version> >>>> <build> >>>> <sourceDirectory>src/main/scala</sourceDirectory> >>>> <plugins> >>>> <plugin> >>>> <!-- see http://davidb.github.com/scala-maven-plugin --> >>>> <groupId>net.alchim31.maven</groupId> >>>> <artifactId>scala-maven-plugin</artifactId> >>>> <version>3.1.3</version> >>>> <executions> >>>> <execution> >>>> <goals> >>>> <goal>compile</goal> >>>> <goal>testCompile</goal> >>>> </goals> >>>> <configuration> >>>> </configuration> >>>> </execution> >>>> </executions> >>>> </plugin> >>>> <plugin> >>>> <groupId>org.apache.maven.plugins</groupId> >>>> <artifactId>maven-surefire-plugin</artifactId> >>>> <version>2.13</version> >>>> <configuration> >>>> <useFile>false</useFile> >>>> <disableXmlReport>true</disableXmlReport> >>>> <!-- If you have classpath issue like >>>> NoDefClassError,... --> >>>> <!-- useManifestOnlyJar>false</useManifestOnlyJar --> >>>> <includes> >>>> <include>**/*Test.*</include> >>>> <include>**/*Suite.*</include> >>>> </includes> >>>> </configuration> >>>> </plugin> >>>> >>>> <!-- "package" command plugin --> >>>> <plugin> >>>> <artifactId>maven-assembly-plugin</artifactId> >>>> <version>2.4.1</version> >>>> <configuration> >>>> <descriptorRefs> >>>> >>>> <descriptorRef>jar-with-dependencies</descriptorRef> >>>> </descriptorRefs> >>>> </configuration> >>>> <executions> >>>> <execution> >>>> <id>make-assembly</id> >>>> <phase>package</phase> >>>> <goals> >>>> <goal>single</goal> >>>> </goals> >>>> </execution> >>>> </executions> >>>> </plugin> >>>> </plugins> >>>> </build> >>>> <dependencies> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-core</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-core</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-clients_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.derby</groupId> >>>> <artifactId>derby</artifactId> >>>> <version>10.13.1.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-jdbc_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-table-api-scala_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-table-api-java</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-table</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-table-planner_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-json</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-scala_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-scala_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-streaming-scala_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-connector-kinesis_2.11</artifactId> >>>> <version>1.8.0</version> >>>> <scope>system</scope> >>>> >>>> <systemPath>${project.basedir}/Jars/flink-connector-kinesis_2.11-1.8-SNAPSHOT.jar</systemPath> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-connector-kafka-0.11_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>amazon-kinesis-client</artifactId> >>>> <version>1.8.8</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>aws-java-sdk-kinesis</artifactId> >>>> <version>1.11.579</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>commons-dbcp</groupId> >>>> <artifactId>commons-dbcp</artifactId> >>>> <version>1.2.2</version> >>>> </dependency> >>>> <dependency> >>>> <groupId>com.google.code.gson</groupId> >>>> <artifactId>gson</artifactId> >>>> <version>2.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>commons-cli</groupId> >>>> <artifactId>commons-cli</artifactId> >>>> <version>1.4</version> >>>> </dependency> >>>> >>>> <!-- >>>> https://mvnrepository.com/artifact/org.apache.commons/commons-csv --> >>>> <dependency> >>>> <groupId>org.apache.commons</groupId> >>>> <artifactId>commons-csv</artifactId> >>>> <version>1.7</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.commons</groupId> >>>> <artifactId>commons-compress</artifactId> >>>> <version>1.4.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>dynamodb-streams-kinesis-adapter</artifactId> >>>> <version>1.4.0</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>dynamodb-streams-kinesis-adapter</artifactId> >>>> <version>1.4.0</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>aws-java-sdk</artifactId> >>>> <version>1.11.579</version> >>>> </dependency> >>>> >>>> >>>> <!-- For Parquet --> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-hadoop-compatibility_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-avro</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> <dependency> >>>> <groupId>org.apache.parquet</groupId> >>>> <artifactId>parquet-avro</artifactId> >>>> <version>1.10.0</version> >>>> </dependency> >>>> <dependency> >>>> <groupId>org.apache.hadoop</groupId> >>>> <artifactId>hadoop-mapreduce-client-core</artifactId> >>>> <version>3.1.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-connector-twitter_2.10</artifactId> >>>> <version>1.1.4-hadoop1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-connector-filesystem_2.11</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.json4s</groupId> >>>> <artifactId>json4s-jackson_2.11</artifactId> >>>> <version>3.6.7</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>com.amazonaws</groupId> >>>> <artifactId>aws-java-sdk-cloudsearch</artifactId> >>>> <version>1.11.500</version> >>>> </dependency> >>>> >>>> <!-- >>>> https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop2 >>>> --> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-shaded-hadoop2</artifactId> >>>> <version>2.8.3-1.8.3</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.flink</groupId> >>>> <artifactId>flink-s3-fs-hadoop</artifactId> >>>> <version>1.8.1</version> >>>> </dependency> >>>> >>>> <dependency> >>>> <groupId>org.apache.hadoop</groupId> >>>> <artifactId>hadoop-common</artifactId> >>>> <version>2.8.5</version> >>>> </dependency> >>>> >>>> >>>> </dependencies> >>>> >>>> </project> >>>> >>>> Scala Code:- >>>> >>>> package com.aws.examples.s3 >>>> >>>> >>>> import org.apache.flink.api.common.typeinfo.Types >>>> import org.apache.flink.api.java.{DataSet, ExecutionEnvironment} >>>> import org.apache.flink.table.api.{Table, TableEnvironment} >>>> import org.apache.flink.table.api.java.BatchTableEnvironment >>>> import org.apache.flink.table.sources.CsvTableSource >>>> >>>> object Batch { >>>> >>>> def main(args: Array[String]): Unit = { >>>> >>>> val env: ExecutionEnvironment = >>>> ExecutionEnvironment.getExecutionEnvironment >>>> val tableEnv: BatchTableEnvironment = >>>> TableEnvironment.getTableEnvironment(env) >>>> /* create table from csv */ >>>> >>>> val tableSrc = CsvTableSource >>>> .builder() >>>> .path("s3a://bucket/csvfolder/avg.txt") >>>> .fieldDelimiter(",") >>>> .field("date", Types.STRING) >>>> .field("month", Types.STRING) >>>> .field("category", Types.STRING) >>>> .field("product", Types.STRING) >>>> .field("profit", Types.INT) >>>> .build() >>>> >>>> tableEnv.registerTableSource("CatalogTable", tableSrc) >>>> >>>> val catalog: Table = tableEnv.scan("CatalogTable") >>>> /* querying with Table API */ >>>> >>>> val order20: Table = catalog >>>> .filter(" category === 'Category5'") >>>> .groupBy("month") >>>> .select("month, profit.sum as sum") >>>> .orderBy("sum") >>>> >>>> val order20Set: DataSet[Row1] = tableEnv.toDataSet(order20, >>>> classOf[Row1]) >>>> >>>> order20Set.writeAsText("src/main/resources/table1/table1") >>>> >>>> //tableEnv.toAppendStream(order20, >>>> classOf[Row]).writeAsText("/home/jivesh/table") >>>> env.execute("State") >>>> >>>> } >>>> >>>> class Row1 { >>>> >>>> var month: String = _ >>>> >>>> var sum: java.lang.Integer = _ >>>> >>>> override def toString(): String = month + "," + sum >>>> >>>> } >>>> >>>> } >>>> >>>> Error:- >>>> *Caused by: >>>> org.apache.flink.fs.s3base.shaded.com.amazonaws.SdkClientException: Unable >>>> to load credentials from service endpoint* >>>> >>>> *Caused by: >>>> org.apache.flink.fs.s3base.shaded.com.amazonaws.AmazonClientException: No >>>> AWS Credentials provided by BasicAWSCredentialsProvider >>>> EnvironmentVariableCredentialsProvider InstanceProfileCredentialsProvider : >>>> org.apache.flink.fs.s3base.shaded.com.amazonaws.SdkClientException: Unable >>>> to load credentials from service endpoint* >>>> >>>> >>>> Thanks >>>> >>>> ------------------------------ >>>> >>>> The information contained in this e-mail is confidential and/or >>>> proprietary to Capital One and/or its affiliates and may only be used >>>> solely in performance of work or services for Capital One. The information >>>> transmitted herewith is intended only for use by the individual or entity >>>> to which it is addressed. If the reader of this message is not the intended >>>> recipient, you are hereby notified that any review, retransmission, >>>> dissemination, distribution, copying or other use of, or taking of any >>>> action in reliance upon this information is strictly prohibited. If you >>>> have received this communication in error, please contact the sender and >>>> delete the material from your computer. >>>> >>>> >>>> >>>> >>>> -- >>> Thanks & Regards >>> Sri Tummala >>> >>> >> >> -- >> Thanks & Regards >> Sri Tummala >> >> > > -- > Thanks & Regards > Sri Tummala > > -- Thanks & Regards Sri Tummala
